diff --git a/AGENTS.md b/AGENTS.md index 318b616..c77f4f3 100644 --- a/AGENTS.md +++ b/AGENTS.md @@ -32,6 +32,7 @@ The `x-media` library: `site::fetch(url)` dispatches (in order) twitter → bsky | `crates/xmedia-bot/src/config.rs` | Manual env parsing into `Config` | | `crates/xmedia-bot/src/handlers.rs` | `Command` enum (teloxide `BotCommands`), message/inline/callback handlers, URL extraction, global statics; per-URL work spawned with a `Semaphore(8)` cap (teloxide's per-chat workers are sequential — batch-forwards need concurrency) | | `crates/xmedia-bot/src/state.rs` | `ChatStore`: parking_lot `Mutex` cache + SQLite write-through (`chat_state` table) | +| `crates/xmedia-bot/src/link_cache.rs` | `LinkCache`: SQLite-backed cache (`link_cache` table) of successfully sent posts — raw caption fields + Telegram `file_id`s; repeat links re-send locally (no fetch/upload), TTL + prune, invalidated on permanent send failure | | `crates/xmedia-bot/src/queue.rs` | `PersistentTaskQueue`: SQLite-backed queue (`tasks` table), `QUEUE_WORKERS = 4` concurrent workers (lease via `BEGIN IMMEDIATE` + `locked_until` TTL), retry→dead-letter, `Notify::notify_waiters` wakeup, `busy_timeout` on all connections | | `crates/xmedia-bot/src/send.rs` | Media senders, upload fallback, error classification, queue task handlers | @@ -81,7 +82,7 @@ Docker: `docker build -t tgxmb .` then `docker run --rm -d --name tgxmb --env-fi - **Rust, stable, edition 2024**, workspace resolver 3. No `rust-version`/MSRV pin, no `rust-toolchain.toml` — recent stable is assumed. No nightly features. - Package manager: **Cargo** (workspace with path dep `x-media` ← `xmedia-bot`). No `[workspace.package]`/shared deps — each crate lists deps independently. - **Two reqwest versions coexist in the lock** (0.12.28 via teloxide, 0.13.3 in x-media) — don't unify casually. -- Config is **environment-variable driven** (dotenv loads `.env`, gitignored; no `.env.example` exists). Key vars: `TELOXIDE_TOKEN` (required), `PIXIV_REFRESH_TOKEN`, `TWITTER_AUTH_TOKEN` (optional; x.com `auth_token` cookie — enables the logged-in GraphQL fallback that fetches NSFW tweets syndication withholds), `BOT_ADMIN` (comma-separated ids), `EDIT_MESSAGE_TTL_SECONDS` (default 86400), `WEBHOOK`/`WEBHOOK_URL`/`WEBHOOK_LISTEN`/`WEBHOOK_PORT`/`WEBHOOK_CERT`/`WEBHOOK_SECRET_TOKEN` (webhook mode requires URL/listen/port, `.expect`ed; `WEBHOOK_CERT` is Telegram-facing self-signed validation only — TLS must be terminated by a reverse proxy), `RUST_LOG`, `TELOXIDE_PROXY`, `LOCAL_USER_ID` (entrypoint only). +- Config is **environment-variable driven** (dotenv loads `.env`, gitignored; no `.env.example` exists). Key vars: `TELOXIDE_TOKEN` (required), `PIXIV_REFRESH_TOKEN`, `TWITTER_AUTH_TOKEN` (optional; x.com `auth_token` cookie — enables the logged-in GraphQL fallback that fetches NSFW tweets syndication withholds), `BOT_ADMIN` (comma-separated ids), `EDIT_MESSAGE_TTL_SECONDS` (default 86400), `LINK_CACHE_TTL_SECONDS` (default 604800), `WEBHOOK`/`WEBHOOK_URL`/`WEBHOOK_LISTEN`/`WEBHOOK_PORT`/`WEBHOOK_CERT`/`WEBHOOK_SECRET_TOKEN` (webhook mode requires URL/listen/port, `.expect`ed; `WEBHOOK_CERT` is Telegram-facing self-signed validation only — TLS must be terminated by a reverse proxy), `RUST_LOG`, `TELOXIDE_PROXY`, `LOCAL_USER_ID` (entrypoint only). - SQLite via `rusqlite` with `bundled` feature (no system libsqlite needed). DB file `data/task_queue.db` is CWD-relative — run from the workspace root, or `/app` in Docker. Mount `./data` and `./cert` volumes. - `.gitattributes` enforces LF for `*.sh` (CRLF breaks shebangs in containers). `.gitignore`: `.env`, `data/`, `cert/`, `docker-compose.yml`, `/target`, `.idea/`. - Docs are in Chinese; user-facing bot strings too. Keep that convention when editing captions/templates/docs. diff --git a/README.md b/README.md index bea059e..dde073c 100644 --- a/README.md +++ b/README.md @@ -10,6 +10,7 @@ Telegram 机器人，将 X / Twitter、Pixiv、Bluesky 的帖子链接转换为 - 可绑定转发频道自动转发；支持转发前编辑 caption 与自定义模板 - 发送失败自动重试并持久化，重试耗尽后通知用户 - Pixiv ugoira 动图自动转码为 MP4 +- 链接结果本地缓存：成功发送后缓存 Telegram file id 与 caption 等，再次收到相同链接直接本地重发，不再请求源站、不保存媒体文件（`LINK_CACHE_TTL_SECONDS` 控制过期，默认 7 天） ## 快速开始 @@ -28,7 +29,7 @@ docker build -t tgxmb . docker run --rm -d --name tgxmb --env-file .env -v ./data:/app/data tgxmb ``` -环境变量：`TELOXIDE_TOKEN`（必填）、`PIXIV_REFRESH_TOKEN`、`BOT_ADMIN`、`EDIT_MESSAGE_TTL_SECONDS`、`RUST_LOG`、`WEBHOOK*`、`TWITTER_AUTH_TOKEN`（可选）。 +环境变量：`TELOXIDE_TOKEN`（必填）、`PIXIV_REFRESH_TOKEN`、`BOT_ADMIN`、`EDIT_MESSAGE_TTL_SECONDS`、`LINK_CACHE_TTL_SECONDS`、`RUST_LOG`、`WEBHOOK*`、`TWITTER_AUTH_TOKEN`（可选）。 NSFW 推文：公开的 syndication 接口不返回敏感内容。设置 `TWITTER_AUTH_TOKEN`（登录 x.com 后浏览器 Cookie 里的 `auth_token` 值）后，bot 会仅在遇到 NSFW 推文时以登录态获取媒体；未设置则提示无媒体。 @@ -60,6 +61,7 @@ Telegram 只接受 443/80/88/8443 端口。 | `PIXIV_REFRESH_TOKEN` | Pixiv 刷新令牌；未设置则禁用 Pixiv | | `BOT_ADMIN` | 管理员聊天 ID，逗号分隔；接收启动/停止通知 | | `EDIT_MESSAGE_TTL_SECONDS` | 转发前编辑记录过期秒数，默认 86400 | +| `LINK_CACHE_TTL_SECONDS` | 链接结果缓存过期秒数，默认 604800（7 天） | | `RUST_LOG` | 日志级别 | | `LOCAL_USER_ID` | 容器内运行用户 UID，默认 9001 | | `VIRTUAL_HOST` | 对外域名或 IP，nginx-proxy 按此路由 | diff --git a/crates/x-media/src/site/mod.rs b/crates/x-media/src/site/mod.rs index 8b883d7..50bae74 100644 --- a/crates/x-media/src/site/mod.rs +++ b/crates/x-media/src/site/mod.rs @@ -68,18 +68,73 @@ impl Fetched { /// caption. pub fn caption_with(&self, format: &str) -> String { match (&self.render_data, format.is_empty()) { - (Some(data), false) => { - let escaped = html_escape::encode_text(format).into_owned(); - escaped - .replace("{url}", &data.url) - .replace("{author}", &data.author) - .replace("{author_url}", &data.author_url) - .replace("{title}", &data.title) - .replace("{tags}", &data.tags) - } + (Some(data), false) => caption_from_fields( + format, + "", + &data.url, + &data.author, + &data.author_url, + &data.title, + &data.tags, + ), _ => self.caption.clone(), } } + + /// The pre-escaped placeholder values (author, author_url, title, tags) + /// a caller needs to rebuild a caption later, e.g. for a cached post + /// where the [`Fetched`] is no longer available. + pub fn render_fields(&self) -> Option<(&str, &str, &str, &str)> { + self.render_data.as_ref().map(|d| { + ( + d.author.as_str(), + d.author_url.as_str(), + d.title.as_str(), + d.tags.as_str(), + ) + }) + } +} + +/// Renders a user-supplied caption format from raw (already-escaped) field +/// values with the same escaping/substitution rules as +/// [`Fetched::caption_with`]. An empty format returns `built_in` unchanged. +pub fn caption_from_fields( + format: &str, + built_in: &str, + url: &str, + author: &str, + author_url: &str, + title: &str, + tags: &str, +) -> String { + if format.is_empty() { + return built_in.to_string(); + } + let escaped = html_escape::encode_text(format).into_owned(); + escaped + .replace("{url}", url) + .replace("{author}", author) + .replace("{author_url}", author_url) + .replace("{title}", title) + .replace("{tags}", tags) +} + +/// Stable per-post cache key derived from any supported URL, so variant +/// domains (x.com / twitter.com / fxtwitter.com, mobile, `/photo/N` +/// suffixes) map to the same post. Returns `"twitter:"`, +/// `"pixiv:"` or `"bsky:/"`. +pub fn cache_key(url: &str) -> Option { + if let Some(caps) = twitter::PATTERN.captures(url) { + return Some(format!("twitter:{}", &caps[1])); + } + if let Some(caps) = pixiv::PATTERN.captures(url) { + return Some(format!("pixiv:{}", &caps[1])); + } + if let Some(caps) = bsky::PATTERN.captures(url) { + return Some(format!("bsky:{}/{}", &caps[1], &caps[2])); + } + None } #[derive(Debug)] @@ -226,6 +281,55 @@ pub async fn download_media(url: &str) -> Result { mod tests { use super::*; + #[test] + fn cache_key_normalizes_domain_variants() { + assert_eq!( + cache_key("https://x.com/user/status/1234567890/photo/1"), + Some("twitter:1234567890".into()) + ); + assert_eq!( + cache_key("https://mobile.twitter.com/user/status/1234567890"), + Some("twitter:1234567890".into()) + ); + assert_eq!( + cache_key("https://fxtwitter.com/user/status/1234567890"), + Some("twitter:1234567890".into()) + ); + assert_eq!( + cache_key("https://www.pixiv.net/artworks/123456"), + Some("pixiv:123456".into()) + ); + assert_eq!( + cache_key("https://bsky.app/profile/handle.example/post/3lorem"), + Some("bsky:handle.example/3lorem".into()) + ); + assert_eq!(cache_key("https://example.com/not-a-post"), None); + } + + #[test] + fn caption_from_fields_substitutes_and_escapes() { + // The format string is escaped, the field values are substituted + // verbatim (callers pass the already-escaped render data). + let out = caption_from_fields( + "see {author} at {url} — {title}", + "", + "https://x.com/u/status/1", + "A & B", + "https://x.com/u", + "hello ", + "", + ); + assert_eq!( + out, + "see A & B at https://x.com/u/status/1 — hello " + ); + // Empty format keeps the built-in caption untouched. + assert_eq!( + caption_from_fields("", "built-in", "u", "a", "au", "t", "g"), + "built-in" + ); + } + #[tokio::test] async fn unsupported_url_returns_none() { let result = fetch("https://example.com/some/article").await; diff --git a/crates/xmedia-bot/src/config.rs b/crates/xmedia-bot/src/config.rs index e8ed7a2..992f376 100644 --- a/crates/xmedia-bot/src/config.rs +++ b/crates/xmedia-bot/src/config.rs @@ -10,6 +10,8 @@ pub struct Config { pub admin_ids: Vec, /// 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, @@ -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::().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, diff --git a/crates/xmedia-bot/src/handlers.rs b/crates/xmedia-bot/src/handlers.rs index 1f1f174..7c670ee 100644 --- a/crates/xmedia-bot/src/handlers.rs +++ b/crates/xmedia-bot/src/handlers.rs @@ -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 = LazyLock::new(|| { }); pub static TASK_QUEUE: LazyLock = LazyLock::new(|| PersistentTaskQueue::new("data/task_queue.db")); +pub static LINK_CACHE: LazyLock = + LazyLock::new(|| LinkCache::open("data/task_queue.db")); pub static CONFIG: LazyLock = 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, + 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, + 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 = 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 = 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 = 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; } } } diff --git a/crates/xmedia-bot/src/link_cache.rs b/crates/xmedia-bot/src/link_cache.rs new file mode 100644 index 0000000..be79cb4 --- /dev/null +++ b/crates/xmedia-bot/src/link_cache.rs @@ -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, +} + +/// 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 { + 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 { + 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> { + 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 { + 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()); + } +} diff --git a/crates/xmedia-bot/src/main.rs b/crates/xmedia-bot/src/main.rs index 984e7f2..6857e64 100644 --- a/crates/xmedia-bot/src/main.rs +++ b/crates/xmedia-bot/src/main.rs @@ -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. diff --git a/crates/xmedia-bot/src/send.rs b/crates/xmedia-bot/src/send.rs index 2aca52d..fd103d2 100644 --- a/crates/xmedia-bot/src/send.rs +++ b/crates/xmedia-bot/src/send.rs @@ -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, + /// `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, #[serde(default)] fallback_url: Option, + /// `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, notify_chat_id: Option, notify_message_id: Option, + /// 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, }, SendAnimation { chat_id: i64, @@ -79,6 +93,10 @@ pub enum Task { forward_channel_id: Option, notify_chat_id: Option, notify_message_id: Option, + /// 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, }, 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 { + 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) { + 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) { + 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 { } } +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 { + 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, 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 = 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, 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, 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, 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, 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, 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, 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, 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"); diff --git a/docker-compose.yml.example b/docker-compose.yml.example index c2b4281..8de6fac 100644 --- a/docker-compose.yml.example +++ b/docker-compose.yml.example @@ -46,6 +46,8 @@ services: # that the public syndication endpoint withholds. TWITTER_AUTH_TOKEN: '' EDIT_MESSAGE_TTL_SECONDS: '86400' + # Link-result cache TTL (default 604800 = 7 days). + LINK_CACHE_TTL_SECONDS: '604800' RUST_LOG: 'info' VIRTUAL_HOST: 'bot.example.com' VIRTUAL_PORT: '8443'