YoursFunny/TelegramTwitterMediaBot
1
0
Fork 0
mirror of https://github.com/TheFunny/TelegramTwitterMediaBot.git synced 2026-09-23 23:32:05 +00:00
Code Releases Activity

feat: cache sent media file ids for instant repeat sends

Browse Source
This commit is contained in:
YoursFunny
2026-08-05 02:06:30 +08:00
parent ab2306002a
commit 51b40cdb42
9 changed files with 726 additions and 92 deletions
Download Patch File Download Diff File Expand all files Collapse all files
crates/xmedia-bot/src/config.rs
+9
View File
@@ -10,6 +10,8 @@ pub struct Config {
pub admin_ids: Vec<i64>,
/// EDIT_MESSAGE_TTL_SECONDS, default 86400 (24h).
pub edit_message_ttl: Duration,
/// LINK_CACHE_TTL_SECONDS, default 604800 (7 days).
pub link_cache_ttl: Duration,
// Webhook settings (moved out of main; names/defaults unchanged).
pub webhook_enabled: bool,
pub webhook_url: Option<url::Url>,
@@ -36,6 +38,12 @@ impl Config {
.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 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());
@@ -53,6 +61,7 @@ impl Config {
Config {
admin_ids,
edit_message_ttl,
link_cache_ttl,
webhook_enabled,
webhook_url,
webhook_listen,
crates/xmedia-bot/src/handlers.rs
+171 -60
View File
@@ -1,7 +1,8 @@
use crate::config::Config;
use crate::link_cache::{CachedMediaKind, CachedPost, LinkCache};
use crate::queue::PersistentTaskQueue;
use crate::send::{self, MediaItemPayload, Task};
use crate::state::{ChatStore, unix_now};
use crate::state::{ChatData, ChatStore, unix_now};
use std::collections::HashSet;
use std::sync::LazyLock;
use teloxide::prelude::*;
@@ -20,6 +21,8 @@ pub static CHAT_STORE: LazyLock<ChatStore> = LazyLock::new(|| {
});
pub static TASK_QUEUE: LazyLock<PersistentTaskQueue> =
LazyLock::new(|| PersistentTaskQueue::new("data/task_queue.db"));
pub static LINK_CACHE: LazyLock<LinkCache> =
LazyLock::new(|| LinkCache::open("data/task_queue.db"));
pub static CONFIG: LazyLock<Config> = LazyLock::new(Config::load);
/// Cap on concurrent per-URL processing. teloxide dispatches updates to a
@@ -339,18 +342,21 @@ fn media_to_payload(media: &Media, sensitive: bool) -> MediaItemPayload {
media: media.url().to_string(),
has_spoiler: sensitive,
fallback_url,
file_id: false,
},
Media::Video { .. } => MediaItemPayload::Video {
media: media.url().to_string(),
has_spoiler: sensitive,
thumbnail: thumbnail_for(media),
fallback_url,
file_id: false,
},
Media::Animated { .. } => MediaItemPayload::Video {
media: media.url().to_string(),
has_spoiler: sensitive,
thumbnail: thumbnail_for(media),
fallback_url,
file_id: false,
},
}
}
@@ -363,11 +369,148 @@ 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) {
let result = match task {
Task::SendAnimation { .. } => send::send_animation(&bot, task).await,
Task::SendMediaSequence { .. } => send::send_media_sequence(&bot, task).await,
Task::ForwardMessages { .. } => unreachable!(),
};
match result {
Ok(message_ids) => {
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 }) => {
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;
}
Err(send::SendError::Permanent {
message: err_message,
task,
}) => {
send::invalidate_cache(&task).await;
log::error!("send for {url} failed permanently: {err_message}");
let _ = reply(bot, message.clone(), format!("Send failed: {err_message}")).await;
}
}
}
/// Builds the send task from ready-made items, sharing the payload shape
/// between the fresh-fetch and link-cache paths.
#[allow(clippy::too_many_arguments)]
fn build_send_task(
chat_data: &ChatData,
message: &Message,
source_url: String,
caption: String,
items: Vec<MediaItemPayload>,
cache_data: Option<CachedPost>,
) -> 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,
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),
cache_data,
}
} else {
Task::SendMediaSequence {
chat_id,
reply_to_message_id: message.id.0 as i64,
caption,
media_batches: send::chunk_media_items(items),
batch_index: 0,
sent_message_ids: vec![],
source_url,
edit_before_forward: chat_data.edit_before_forward,
forward_channel_id: chat_data.forward_channel_id,
notify_chat_id: Some(chat_id),
notify_message_id: Some(message.id.0 as i64),
cache_data,
}
}
}
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 {
log::error!("send_chat_action failed: {e}");
}
// Link cache: a post sent before is re-sent from Telegram file ids —
// no source-site request, no download, no upload. Keyed by the
// normalized post id so x.com / fxtwitter / /photo/N variants collide.
if let Some(key) = x_media::site::cache_key(url)
&& let Some(cached) = LINK_CACHE.get(&key, CONFIG.link_cache_ttl).await
{
log::info!("link cache hit for {url}");
let chat_data = CHAT_STORE.get(chat_id).await;
let site = key.split(':').next().unwrap_or("unknown");
let format = chat_data
.message_format
.get(site)
.cloned()
.unwrap_or_default();
let caption = if format.is_empty() {
cached.caption.clone()
} else {
x_media::site::caption_from_fields(
&format,
"",
&cached.url,
&cached.author,
&cached.author_url,
&cached.title,
&cached.tags,
)
};
let items: Vec<MediaItemPayload> = cached
.media
.iter()
.map(|m| match m.kind {
CachedMediaKind::Photo => MediaItemPayload::Photo {
media: m.file_id.clone(),
has_spoiler: cached.sensitive,
fallback_url: None,
file_id: true,
},
CachedMediaKind::Video => MediaItemPayload::Video {
media: m.file_id.clone(),
has_spoiler: cached.sensitive,
thumbnail: None,
fallback_url: None,
file_id: true,
},
CachedMediaKind::Animation => MediaItemPayload::Animation {
media: m.file_id.clone(),
has_spoiler: cached.sensitive,
file_id: true,
},
})
.collect();
let task = build_send_task(
&chat_data,
message,
cached.url.clone(),
caption,
items,
Some(cached),
);
dispatch_send(bot, message, &task, url).await;
return;
}
log::info!("fetching {url}");
match x_media::site::fetch(url).await {
// Unsupported links are ignored silently (Python parity).
@@ -397,66 +540,34 @@ async fn url_media(bot: Bot, message: &Message, url: &str) {
.cloned()
.unwrap_or_default();
let caption = fetched.caption_with(&format);
let task = if fetched.media.len() == 1
&& matches!(fetched.media[0], Media::Animated { .. })
{
Task::SendAnimation {
chat_id,
reply_to_message_id: message.id.0 as i64,
caption: caption.clone(),
animation: MediaItemPayload::Animation {
media: fetched.media[0].url().to_string(),
has_spoiler: fetched.sensitive,
},
source_url: fetched.source_url.clone(),
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),
// Raw render data for the link cache; the send fills in the
// Telegram file ids and persists the entry.
let cache_data = fetched.render_fields().map(|(author, author_url, title, tags)| {
CachedPost {
url: fetched.source_url.clone(),
caption: fetched.caption.clone(),
title: title.to_string(),
author: author.to_string(),
author_url: author_url.to_string(),
tags: tags.to_string(),
sensitive: fetched.sensitive,
media: vec![],
}
} else {
let items: Vec<MediaItemPayload> = fetched
.media
.iter()
.map(|media| media_to_payload(media, fetched.sensitive))
.collect();
Task::SendMediaSequence {
chat_id,
reply_to_message_id: message.id.0 as i64,
caption: caption.clone(),
media_batches: send::chunk_media_items(items),
batch_index: 0,
sent_message_ids: vec![],
source_url: fetched.source_url.clone(),
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),
}
};
let result = match &task {
Task::SendAnimation { .. } => send::send_animation(&bot, &task).await,
Task::SendMediaSequence { .. } => send::send_media_sequence(&bot, &task).await,
Task::ForwardMessages { .. } => unreachable!(),
};
match result {
Ok(message_ids) => {
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 }) => {
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;
}
Err(send::SendError::Permanent {
message: err_message,
..
}) => {
log::error!("send for {url} failed permanently: {err_message}");
let _ = reply(bot, message.clone(), format!("Send failed: {err_message}")).await;
}
}
});
let items: Vec<MediaItemPayload> = fetched
.media
.iter()
.map(|media| media_to_payload(media, fetched.sensitive))
.collect();
let task = build_send_task(
&chat_data,
message,
fetched.source_url.clone(),
caption,
items,
cache_data,
);
dispatch_send(bot, message, &task, url).await;
}
}
}
crates/xmedia-bot/src/link_cache.rs
+230
View File
@@ -0,0 +1,230 @@
//! Persistent cache of successfully sent posts.
//!
//! After a media send succeeds, the raw render data plus the Telegram
//! `file_id`s of the sent items are stored keyed by [`crate::site` cache
//! key]. A repeated link is then answered entirely from local state — no
//! re-fetch of the source site, no re-upload — and no media file is stored
//! on disk (the file ids point at Telegram's servers). Entries expire after
//! [`Config::link_cache_ttl`]; a stale entry is dropped lazily on read and
//! by the periodic prune in `main`.
use rusqlite::{params, Connection};
use serde::{Deserialize, Serialize};
use std::time::Duration;
#[derive(Serialize, Deserialize, Clone, Debug, PartialEq)]
#[serde(rename_all = "snake_case")]
pub enum CachedMediaKind {
Photo,
Video,
Animation,
}
#[derive(Serialize, Deserialize, Clone, Debug)]
pub struct CachedMedia {
pub kind: CachedMediaKind,
pub file_id: String,
}
/// Everything needed to re-send a post without touching the source site:
/// the canonical URL, pre-escaped caption fields, and the file ids produced
/// by the original successful send.
#[derive(Serialize, Deserialize, Clone, Debug)]
pub struct CachedPost {
pub url: String,
/// The site's built-in caption (used when the chat has no format
/// override).
pub caption: String,
pub title: String,
pub author: String,
pub author_url: String,
pub tags: String,
pub sensitive: bool,
pub media: Vec<CachedMedia>,
}
/// SQLite-backed cache sharing `data/task_queue.db` with the queue and chat
/// state (same `open_db` pattern: busy timeout, `spawn_blocking` I/O).
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)
&& let Err(e) = conn.execute_batch(
"CREATE TABLE IF NOT EXISTS link_cache (url TEXT PRIMARY KEY, \
payload TEXT NOT NULL, created_at REAL NOT NULL);",
)
{
log::error!("failed to initialize link cache schema: {e}");
}
Self {
db_path: db_path.to_string(),
}
}
/// 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 mut stmt =
conn.prepare("SELECT payload, created_at FROM link_cache WHERE url = ?1")?;
let mut rows = stmt.query(params![key])?;
let Some(row) = rows.next()? else {
return Ok(None);
};
let payload: String = row.get(0)?;
let created_at: f64 = row.get(1)?;
if now_f64() - created_at > ttl {
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
})
}
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)?;
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}"));
}
/// 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)?;
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}"));
}
/// 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)?;
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
})
}
}
fn now_f64() -> f64 {
std::time::SystemTime::now()
.duration_since(std::time::UNIX_EPOCH)
.map(|d| d.as_secs_f64())
.unwrap_or(0.0)
}
#[cfg(test)]
mod tests {
use super::*;
fn 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: true,
media: vec![CachedMedia {
kind: CachedMediaKind::Photo,
file_id: "AgAC...".into(),
}],
}
}
#[tokio::test]
async fn put_get_roundtrip() {
let dir = tempfile::tempdir().unwrap();
let cache = LinkCache::open(dir.path().join("c.db").to_str().unwrap());
cache.put("twitter:1", &entry()).await;
let got = cache.get("twitter:1", Duration::from_secs(3600)).await;
assert!(got.is_some());
let got = got.unwrap();
assert_eq!(got.url, "https://x.com/u/status/1");
assert_eq!(got.media[0].file_id, "AgAC...");
}
#[tokio::test]
async fn expired_entry_removed_on_read() {
let dir = tempfile::tempdir().unwrap();
let cache = LinkCache::open(dir.path().join("c.db").to_str().unwrap());
cache.put("twitter:1", &entry()).await;
// Force the row into the past so a 1s TTL expires it.
{
let conn = Connection::open(dir.path().join("c.db")).unwrap();
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());
}
#[tokio::test]
async fn remove_and_prune() {
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;
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());
{
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());
}
}
crates/xmedia-bot/src/main.rs
+6 -1
View File
@@ -9,11 +9,12 @@ use x_media::site;
mod config;
mod handlers;
mod link_cache;
mod queue;
mod send;
mod state;
use handlers::{CHAT_STORE, CONFIG, TASK_QUEUE};
use handlers::{CHAT_STORE, CONFIG, LINK_CACHE, TASK_QUEUE};
/// Docker `stop` / `compose down` delivers SIGTERM, which teloxide's ctrlc
/// handler (SIGINT only) never sees — without this the process would die
@@ -85,6 +86,10 @@ async fn main() {
}
let ttl = CONFIG.edit_message_ttl;
let removed = CHAT_STORE.prune_expired(ttl).await;
let pruned = LINK_CACHE.prune(CONFIG.link_cache_ttl).await;
if pruned > 0 {
log::info!("link cache: pruned {pruned} expired entr(ies)");
}
for (chat_id, prompt_message_id) in removed {
// If the prompt was already deleted, this fails with a
// 400 "message to edit not found" — log and ignore.
crates/xmedia-bot/src/send.rs
+190 -20
View File
@@ -3,7 +3,8 @@
//! URL is blocked by hotlink protection; the bot downloads the file itself
//! and uploads it via multipart).
use crate::handlers::{CHAT_STORE, TASK_QUEUE};
use crate::handlers::{CHAT_STORE, LINK_CACHE, TASK_QUEUE};
use crate::link_cache::{CachedMedia, CachedMediaKind, CachedPost};
use crate::queue::QueueError;
use crate::state::{EditMessage, unix_now};
use rand::Rng;
@@ -29,6 +30,9 @@ pub enum MediaItemPayload {
/// size limits.
#[serde(default)]
fallback_url: Option<String>,
/// `media` is a Telegram file id (link-cache hit), not a URL.
#[serde(default)]
file_id: bool,
},
Video {
media: String,
@@ -36,10 +40,16 @@ pub enum MediaItemPayload {
thumbnail: Option<String>,
#[serde(default)]
fallback_url: Option<String>,
/// `media` is a Telegram file id (link-cache hit), not a URL.
#[serde(default)]
file_id: bool,
},
Animation {
media: String,
has_spoiler: bool,
/// `media` is a Telegram file id (link-cache hit), not a URL.
#[serde(default)]
file_id: bool,
},
}
@@ -68,6 +78,10 @@ pub enum Task {
forward_channel_id: Option<i64>,
notify_chat_id: Option<i64>,
notify_message_id: Option<i64>,
/// Raw render data captured on a cache miss; the send fills in the
/// Telegram file ids and persists the entry (see `link_cache`).
#[serde(default)]
cache_data: Option<CachedPost>,
},
SendAnimation {
chat_id: i64,
@@ -79,6 +93,10 @@ pub enum Task {
forward_channel_id: Option<i64>,
notify_chat_id: Option<i64>,
notify_message_id: Option<i64>,
/// Raw render data captured on a cache miss; the send fills in the
/// Telegram file id and persists the entry (see `link_cache`).
#[serde(default)]
cache_data: Option<CachedPost>,
},
ForwardMessages {
from_chat_id: i64,
@@ -89,6 +107,108 @@ 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::ForwardMessages { .. } => None,
}
}
fn source_url(&self) -> Option<&str> {
match self {
Task::SendMediaSequence { source_url, .. }
| Task::SendAnimation { source_url, .. } => Some(source_url),
Task::ForwardMessages { .. } => None,
}
}
/// True when the media payloads are Telegram file ids from the link cache
/// (a cached file id that goes permanently bad should be dropped so the
/// next request re-fetches).
fn is_cached_send(&self) -> bool {
self.cache_data().is_some_and(|c| !c.media.is_empty())
}
}
/// Telegram file id of the message's media, matched to the payload kind.
fn file_id_of_message(message: &Message, item: &MediaItemPayload) -> Option<String> {
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::Video { .. } => message.video().map(|v| v.file.id.to_string()),
MediaItemPayload::Animation { .. } => message.animation().map(|a| a.file.id.to_string()),
}
}
fn kind_of_item(item: &MediaItemPayload) -> CachedMediaKind {
match item {
MediaItemPayload::Photo { .. } => CachedMediaKind::Photo,
MediaItemPayload::Video { .. } => CachedMediaKind::Video,
MediaItemPayload::Animation { .. } => CachedMediaKind::Animation,
}
}
/// Collects the Telegram file ids of a sent media group, aligned to the
/// batch's items.
fn collect_file_ids(messages: &[Message], batch: &[MediaItemPayload], out: &mut Vec<CachedMedia>) {
for (message, item) in messages.iter().zip(batch.iter()) {
if let Some(file_id) = file_id_of_message(message, item) {
out.push(CachedMedia {
kind: kind_of_item(item),
file_id,
});
}
}
}
/// Persists a successful send under the post's cache key. Only runs for a
/// fresh (non-resumed) task that carried raw cache data with no file ids yet.
async fn cache_sent_task(task: &Task, media: Vec<CachedMedia>) {
let Some(cache_data) = task.cache_data() else {
return;
};
if !cache_data.media.is_empty() || media.is_empty() {
return;
}
let mut post = cache_data.clone();
post.media = media;
if let Some(key) = x_media::site::cache_key(&post.url) {
LINK_CACHE.put(&key, &post).await;
log::info!("cached send for {}", post.url);
}
}
/// Persists a lone animation send under the post's cache key.
async fn cache_animation_send(task: &Task, message: &Message) {
if let Some(file_id) = message.animation().map(|a| a.file.id.to_string()) {
cache_sent_task(
task,
vec![CachedMedia {
kind: CachedMediaKind::Animation,
file_id,
}],
)
.await;
}
}
/// 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) {
if task.is_cached_send()
&& let Some(url) = task.source_url()
&& let Some(key) = x_media::site::cache_key(url)
{
log::info!("removing stale link cache entry for {url}");
LINK_CACHE.remove(&key).await;
}
}
pub const MAX_MEDIA_GROUP: usize = 9;
/// Upload cap (bytes): files above this are not uploaded; the bot falls back
/// to a smaller media URL instead.
@@ -198,6 +318,32 @@ fn input_file_for(media: &str) -> Result<InputFile, String> {
}
}
impl MediaItemPayload {
/// The input for a send: a cached file id goes out as `InputFile::file_id`
/// (no fetch, no upload), URLs go to Telegram, anything else is a local
/// path (transient upload fallback).
fn input_file(&self) -> Result<InputFile, String> {
match self {
MediaItemPayload::Photo {
media,
file_id: true,
..
}
| MediaItemPayload::Video {
media,
file_id: true,
..
}
| MediaItemPayload::Animation {
media,
file_id: true,
..
} => Ok(InputFile::file_id(media.clone().into())),
_ => input_file_for(item_url(self)),
}
}
}
fn photo_media(file: InputFile, caption: Option<&str>, spoiler: bool) -> InputMedia {
let mut photo = InputMediaPhoto::new(file).parse_mode(ParseMode::Html);
if let Some(caption) = caption {
@@ -244,27 +390,22 @@ fn build_media_group(
let item_caption = if i == 0 { caption } else { None };
Ok(match item {
MediaItemPayload::Photo {
media,
has_spoiler,
..
} => photo_media(input_file_for(media)?, item_caption, *has_spoiler),
has_spoiler, ..
} => photo_media(item.input_file()?, item_caption, *has_spoiler),
MediaItemPayload::Video {
media,
has_spoiler,
thumbnail,
..
} => {
let mut video = video_media(input_file_for(media)?, item_caption, *has_spoiler);
let mut video = video_media(item.input_file()?, item_caption, *has_spoiler);
if let (Some(thumb), InputMedia::Video(v)) = (thumbnail, &mut video) {
*v = v.clone().thumbnail(input_file_for(thumb)?);
}
video
}
MediaItemPayload::Animation {
media,
has_spoiler,
..
} => animation_media(input_file_for(media)?, item_caption, *has_spoiler),
has_spoiler, ..
} => animation_media(item.input_file()?, item_caption, *has_spoiler),
})
})
.collect()
@@ -459,12 +600,14 @@ fn updated_sequence_task(task: &Task, batch_index: usize, sent_message_ids: Vec<
reply_to_message_id,
caption,
media_batches,
batch_index: _,
sent_message_ids: _,
source_url,
edit_before_forward,
forward_channel_id,
notify_chat_id,
notify_message_id,
..
cache_data,
} => Task::SendMediaSequence {
chat_id: *chat_id,
reply_to_message_id: *reply_to_message_id,
@@ -477,6 +620,7 @@ fn updated_sequence_task(task: &Task, batch_index: usize, sent_message_ids: Vec<
forward_channel_id: *forward_channel_id,
notify_chat_id: *notify_chat_id,
notify_message_id: *notify_message_id,
cache_data: cache_data.clone(),
},
_ => unreachable!("updated_sequence_task requires a SendMediaSequence task"),
}
@@ -501,6 +645,10 @@ pub async fn send_media_sequence(bot: &Bot, task: &Task) -> Result<Vec<i64>, Sen
let chat_id = *chat_id;
let reply_to = *reply_to_message_id;
let mut sent = sent_message_ids.clone();
// File ids accumulated across batches for the link cache. Only a fresh
// (non-resumed) full send populates the cache.
let mut cached_media: Vec<CachedMedia> = Vec::new();
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 };
@@ -524,6 +672,7 @@ pub async fn send_media_sequence(bot: &Bot, task: &Task) -> Result<Vec<i64>, Sen
media_batches.len(),
batch.len()
);
collect_file_ids(&messages, batch, &mut cached_media);
sent.extend(messages.into_iter().map(|m| m.id.0 as i64));
}
Err(RequestError::Api(api))
@@ -531,13 +680,13 @@ pub async fn send_media_sequence(bot: &Bot, task: &Task) -> Result<Vec<i64>, Sen
{
log::info!(
"Telegram could not fetch media for batch {idx} ({}), downloading and reuploading",
batch
.first()
.map(|item| item_url(item))
.unwrap_or("?")
batch.first().map(item_url).unwrap_or("?")
);
match send_batch_via_upload(bot, chat_id, reply_to, batch, caption).await {
Ok(messages) => sent.extend(messages.into_iter().map(|m| m.id.0 as i64)),
Ok(messages) => {
collect_file_ids(&messages, batch, &mut cached_media);
sent.extend(messages.into_iter().map(|m| m.id.0 as i64));
}
Err(FallbackError::Retryable { delay_seconds }) => {
return Err(SendError::Retryable {
delay_seconds,
@@ -561,6 +710,9 @@ pub async fn send_media_sequence(bot: &Bot, task: &Task) -> Result<Vec<i64>, Sen
}
}
}
if fresh_send {
cache_sent_task(task, cached_media).await;
}
Ok(sent)
}
@@ -601,6 +753,7 @@ pub async fn send_animation(bot: &Bot, task: &Task) -> Result<Vec<i64>, SendErro
MediaItemPayload::Animation {
media,
has_spoiler,
..
} => (media, *has_spoiler),
MediaItemPayload::Photo { .. } | MediaItemPayload::Video { .. } => {
unreachable!("SendAnimation carries an Animation payload")
@@ -613,7 +766,11 @@ pub async fn send_animation(bot: &Bot, task: &Task) -> Result<Vec<i64>, SendErro
match send_animation_inner(bot, chat_id, reply_to, caption, has_spoiler, url_file)
.await
{
Ok(message) => Ok(vec![message.id.0 as i64]),
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) =>
{
@@ -634,7 +791,11 @@ pub async fn send_animation(bot: &Bot, task: &Task) -> Result<Vec<i64>, SendErro
)
.await
{
Ok(message) => Ok(vec![message.id.0 as i64]),
Ok(message) => {
let id = message.id.0 as i64;
cache_animation_send(task, &message).await;
Ok(vec![id])
}
Err(e) => Err(classify_to_send_error(&e, task.clone())),
}
}
@@ -652,7 +813,11 @@ pub async fn send_animation(bot: &Bot, task: &Task) -> Result<Vec<i64>, SendErro
)
.await
{
Ok(message) => Ok(vec![message.id.0 as i64]),
Ok(message) => {
let id = message.id.0 as i64;
cache_animation_send(task, &message).await;
Ok(vec![id])
}
Err(e) => Err(classify_to_send_error(&e, task.clone())),
}
}
@@ -868,6 +1033,7 @@ pub async fn handle_task(payload: serde_json::Value) -> Result<(), QueueError> {
});
}
Err(SendError::Permanent { message, task }) => {
invalidate_cache(&task).await;
return Err(QueueError::Permanent {
message,
payload: serde_json::to_value(task).expect("task serializes"),
@@ -1026,12 +1192,14 @@ mod tests {
media: "https://a/b.jpg".into(),
has_spoiler: true,
fallback_url: Some("https://a/b_small.jpg".into()),
file_id: false,
}],
vec![MediaItemPayload::Video {
media: "https://a/v.mp4".into(),
has_spoiler: false,
thumbnail: Some("https://a/t.jpg".into()),
fallback_url: None,
file_id: false,
}],
],
batch_index: 1,
@@ -1041,6 +1209,7 @@ mod tests {
forward_channel_id: Some(333),
notify_chat_id: Some(111),
notify_message_id: Some(222),
cache_data: None,
};
let json = serde_json::to_value(&task).unwrap();
assert_eq!(json["type"], "send_media_sequence");
@@ -1070,6 +1239,7 @@ mod tests {
media: "https://a/b.jpg".into(),
has_spoiler: false,
fallback_url: None,
file_id: false,
};
let json = serde_json::to_value(&photo).unwrap();
assert_eq!(json["kind"], "photo");