diff --git a/AGENTS.md b/AGENTS.md index b63c098..e5fb567 100644 --- a/AGENTS.md +++ b/AGENTS.md @@ -37,8 +37,9 @@ The `x-media` library: `site::fetch(url)` dispatches through the `SITES` registr | `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_one` worker wakeup plus a separate `Notify` for the 30 s lease-expiry sweep (a shared one let the sweep steal the workers' wakeup permit), `busy_timeout` on all connections | -| `crates/xmedia-bot/src/send.rs` | Media senders, upload fallback, error classification, queue task handlers | -| `crates/xmedia-bot/src/media_sender.rs` | `MediaSender` trait: the send surface (`send_media_group`/`send_animation`/`copy_messages`/`send_message`/`send_chat_action`) implemented by teloxide `Bot` (per-chat rate-limited) and by a scripted `MockSender` in tests | +| `crates/xmedia-bot/src/ctx.rs` | `AppContext`: the injected collaborators (`sender` + `ChatStore`/`PersistentTaskQueue`/`LinkCache`/`Config`), `from_statics` for production and the `CONTEXT` static the worker closures hold. `test_support::TestStores` backs handler tests with a tempdir store set | +| `crates/xmedia-bot/src/send/` | `send/mod.rs`: `Task`/`MediaItemPayload` payloads, `SendError`/`Classification`, `send_media_sequence`/`send_animation`/`forward_messages`; `send/input_media.rs`: payload → `InputFile`/`InputMedia` + `build_media_group` (caption on the first item only); `send/upload.rs`: the download-and-reupload fallback (`prepare_upload_item`/`send_batch_via_upload`, photo downscale handoff); `send/post_send.rs`: link-cache write, `KEEP_ALIVE` registry, `settle_task`, `post_send_actions`, `handle_task`/`dead_letter_notify` | +| `crates/xmedia-bot/src/media_sender.rs` | `MediaSender` trait: the user-flow surface (`send_media_group`/`send_animation`/`copy_messages`/`send_message`/`answer_callback_query`/`edit_message_caption`/`delete_message`/`send_chat_action`) implemented by teloxide `Bot` (per-chat rate-limited) and by a recording `MockSender` in tests. Admin/setup APIs (`get_chat`, `set_my_commands`, …) stay on the concrete `Bot` | | `crates/xmedia-bot/src/rate_limit.rs` | Per-chat token bucket (`CAPACITY = 20`, ~20 msg/min refill) paced before sends reach the API so batch forwards don't trip flood control | ## Development Commands @@ -72,8 +73,8 @@ Docker: `docker build -t tgxmb .` then `docker run --rm -d --name tgxmb --env-fi | File | Why it matters | |---|---| | `crates/xmedia-bot/src/main.rs` | Startup sequence, webhook vs polling, graceful shutdown (SIGINT via teloxide ctrlc / SIGTERM via `stop_token` for docker, → sweep stop → admin msg → queue stop) | -| `crates/xmedia-bot/src/handlers/` | `statics.rs` = `CHAT_STORE`/`TASK_QUEUE`/`CONFIG` singletons (open `$DATA_DIR/task_queue.db`, default `data/` **relative to CWD**, dir auto-created); `commands.rs` = command dispatch (incl. the `/test ` parse-only debug command and the admin-only `/bot_dict` state dump); `urls.rs` = URL extraction + the per-URL pipeline (`enqueue_retry` lives in `send.rs`); `inline.rs`/`callback.rs` = inline queries / edit-before-forward buttons | -| `crates/xmedia-bot/src/send.rs` | Constants `MAX_MEDIA_GROUP = 9`; fallback chain; `classify_request_error`; download-and-reupload fallback triggered only by Telegram API errors (`is_media_fetch_failure` / `is_size_error`) | +| `crates/xmedia-bot/src/handlers/` | `statics.rs` = `CHAT_STORE`/`TASK_QUEUE`/`CONFIG` singletons (open `$DATA_DIR/task_queue.db`, default `data/` **relative to CWD**, dir auto-created); `commands.rs` = command dispatch (incl. the `/test ` parse-only debug command and the admin-only `/bot_dict` state dump); `urls.rs` = URL extraction + the per-URL pipeline (`enqueue_retry` lives in `send/post_send.rs`); `inline.rs` = debounced inline queries; `callback.rs` = edit-before-forward buttons (dptree entry + testable `handle_callback` core) | +| `crates/xmedia-bot/src/send/` | `mod.rs`: constants `MAX_MEDIA_GROUP = 9`; `classify_request_error`; the senders. `upload.rs`: download-and-reupload fallback triggered only by Telegram API errors (`is_media_fetch_failure` / `is_size_error`). `post_send.rs`: settlement (`settle_task`), cache write, post-send actions, queue handlers. `input_media.rs`: payload → `InputMedia` | | `crates/xmedia-bot/src/photo.rs` | Pure-Rust photo processing (no ffmpeg): `png` (image-png) decode/encode + `zune-jpeg` decode + `fast_image_resize` Lanczos3 downscale + `jpeg-encoder`. Photos over Telegram's limits (width + height > 10000 px → `PHOTO_INVALID_DIMENSIONS`; bytes > 10 MiB) are decoded, downscaled keeping the format, PNG bit depth > 24 (RGBA 32-bit / 16-bit per channel) reduced to 24-bit RGB with alpha flattened white (≤24-bit untouched, never upconverted), and transcoded to JPEG only if still over the cap; memory budget guarded, otherwise the item's smaller fallback URL | | `crates/x-media/src/site/mod.rs` | Dispatcher, `Fetched`/`FetchError`, shared `CLIENT`, `download_media` (adds `Referer: https://www.pixiv.net/` for `pximg.net` hotlink protection) | | `crates/x-media/src/site/pixiv/api.rs` | OAuth token exchange (hardcoded app client id/secret), access-token cache, ugoira zip→MP4 via ffmpeg in `spawn_blocking` | @@ -101,5 +102,5 @@ Docker: `docker build -t tgxmb .` then `docker run --rm -d --name tgxmb --env-fi - Live-network tests exist in `site/twitter/interface.rs` (5), `site/bsky/interface.rs` (2), `site/misskey/interface.rs` (1), `site/pixiv/api.rs` (1); `photo.rs` adds one `#[ignore = "heavy: …"]` test. `site/mod.rs` also has a **token-gated but not `#[ignore]`d** pixiv download test (`download_media_pixiv_original_with_referer`): it hits `i.pximg.net` whenever `PIXIV_REFRESH_TOKEN` is set, so a local `cargo test --workspace` is not fully offline and can flake on a pixiv CDN body timeout. Test gating convention (enforced by `.github/workflows/ci.yml`): pure unit tests always run; live-network tests carry `#[ignore = "live network: ..."]` (run via `cargo test --workspace -- --ignored live`); token-gated pixiv tests early-return when `PIXIV_REFRESH_TOKEN` is absent **or empty** (an unset GitHub secret arrives as `""` — `is_err()` alone would run them tokenless and fail). Run the full offline suite with `cargo test --workspace`. - Fixtures are inline `serde_json::json!` builder fns (`fixture()`, `thread_json()`, `illust_json()`), not files. The shared `CLIENT` sets `pool_max_idle_per_host(0)` under `#[cfg(test)]` to avoid cross-runtime `DispatchGone`. - **CI** — `.github/workflows/ci.yml` runs `cargo fmt --check` + `cargo clippy --workspace --all-targets -- -D warnings` + `cargo test --workspace` + a `actions-rust-lang/audit` dependency-vulnerability gate (offline, no secrets, on every push/PR) and a `live` job (schedule/manual/tag only, `PIXIV_REFRESH_TOKEN`/`TWITTER_AUTH_TOKEN` from secrets, `continue-on-error`) for the `#[ignore]`d live + token tests. `.github/workflows/docker.yml` builds/pushes the image only. -- Untested and hard to test without a mock seam: `main.rs`, `config.rs`, `db.rs`, `media_sender.rs` (holds `MockSender` itself); `handlers/mod.rs`, `handlers/callback.rs`, `handlers/statics.rs` (need a real teloxide `Bot`); `media.rs`, `lib.rs`, all `model.rs`. `handlers/urls.rs`/`handlers/commands.rs`/`send.rs`/`state.rs`/`queue.rs`/`link_cache.rs`/`rate_limit.rs` are covered through their injected stores and the scripted `MockSender`. +- Untested and hard to test without a mock seam: `main.rs`, `config.rs`, `db.rs`, `handlers/statics.rs`, `media_sender.rs` (holds the `MockSender` itself); in `x-media`: `media.rs`, `lib.rs`, all `model.rs`. The `commands.rs` *executor* needs a real `Bot` (only its pure report builder is tested). Everything else — `handlers/{mod,callback,inline,urls}.rs`, `send/*`, `ctx.rs`, `state.rs`, `queue.rs`, `link_cache.rs`, `rate_limit.rs` — is driven through `TestStores`/`ctx::test_support` and the scripted `MockSender`. - No coverage tracking. diff --git a/crates/xmedia-bot/src/send/input_media.rs b/crates/xmedia-bot/src/send/input_media.rs new file mode 100644 index 0000000..1b65735 --- /dev/null +++ b/crates/xmedia-bot/src/send/input_media.rs @@ -0,0 +1,128 @@ +//! Payload → Telegram input types: `InputFile` selection (cached file id / +//! URL / local path), the per-kind `InputMedia` builders and the media-group +//! assembly with its caption rule. + +use super::MediaItemPayload; +use teloxide::types::{ + InputFile, InputMedia, InputMediaAnimation, InputMediaPhoto, InputMediaVideo, ParseMode, +}; + +fn parse_media_url(s: &str) -> Result { + url::Url::parse(s).map_err(|e| format!("invalid media URL: {e}")) +} + +pub(super) fn item_url(item: &MediaItemPayload) -> &str { + match item { + MediaItemPayload::Photo { media, .. } + | MediaItemPayload::Video { media, .. } + | MediaItemPayload::Animation { media, .. } => media, + } +} + +/// Remote http(s) URLs are handed to Telegram to fetch; everything else +/// (e.g. a locally encoded ugoira MP4) is uploaded directly. +pub(super) fn input_file_for(media: &str) -> Result { + if media.starts_with("http://") || media.starts_with("https://") { + Ok(InputFile::url(parse_media_url(media)?)) + } else if !std::path::Path::new(media).exists() { + // A retried task may reference a temp file the original send's + // TempDir already cleaned up; fail fast and permanent instead of + // burning retries on a file that can never come back. + Err(format!("local media file missing: {media}")) + } else { + Ok(InputFile::file(media)) + } +} + +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)), + } + } +} + +pub(super) 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 { + photo = photo.caption(caption); + } + if spoiler { + photo = photo.spoiler(); + } + InputMedia::Photo(photo) +} + +pub(super) fn video_media(file: InputFile, caption: Option<&str>, spoiler: bool) -> InputMedia { + let mut video = InputMediaVideo::new(file).parse_mode(ParseMode::Html); + if let Some(caption) = caption { + video = video.caption(caption); + } + if spoiler { + video = video.spoiler(); + } + InputMedia::Video(video) +} + +pub(super) fn animation_media(file: InputFile, caption: Option<&str>, spoiler: bool) -> InputMedia { + let mut animation = InputMediaAnimation::new(file).parse_mode(ParseMode::Html); + if let Some(caption) = caption { + animation = animation.caption(caption); + } + if spoiler { + animation = animation.spoiler(); + } + InputMedia::Animation(animation) +} + +/// Builds a media group from payloads; only the first item of the batch gets +/// the caption (Telegram rejects captions on later items). +pub(super) fn build_media_group( + batch: &[MediaItemPayload], + caption: Option<&str>, +) -> Result, String> { + batch + .iter() + .enumerate() + .map(|(i, item)| { + let item_caption = if i == 0 { caption } else { None }; + Ok(match item { + MediaItemPayload::Photo { has_spoiler, .. } => { + photo_media(item.input_file()?, item_caption, *has_spoiler) + } + MediaItemPayload::Video { + has_spoiler, + thumbnail, + .. + } => { + 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 { has_spoiler, .. } => { + animation_media(item.input_file()?, item_caption, *has_spoiler) + } + }) + }) + .collect() +} diff --git a/crates/xmedia-bot/src/send.rs b/crates/xmedia-bot/src/send/mod.rs similarity index 60% rename from crates/xmedia-bot/src/send.rs rename to crates/xmedia-bot/src/send/mod.rs index 4a63d91..abafe91 100644 --- a/crates/xmedia-bot/src/send.rs +++ b/crates/xmedia-bot/src/send/mod.rs @@ -1,28 +1,36 @@ -//! Typed task payloads and send/forward executors with retry classification -//! and the download-and-reupload fallback (Telegram's own fetch of a media -//! URL is blocked by hotlink protection; the bot downloads the file itself -//! and uploads it via multipart). +//! Task payloads and the send/forward executors, split by concern: +//! [`input_media`] builds Telegram input types from payloads, [`upload`] is +//! the download-and-reupload fallback (Telegram's own fetch of a media URL is +//! blocked by hotlink protection, so the bot downloads the file itself and +//! uploads it via multipart), [`post_send`] covers everything around a send +//! (cache write, keep-alive media, settlement, post-send actions, queue +//! entry points). This module keeps the payload types, the error +//! classification and the senders themselves. + +mod input_media; +mod post_send; +mod upload; use crate::ctx::AppContext; -use crate::db::{now_f64, unix_now}; use crate::handlers::log_key; -use crate::link_cache::{CachedMedia, CachedMediaKind, CachedPost, LinkCache}; +use crate::link_cache::{CachedMedia, CachedMediaKind, CachedPost}; use crate::media_sender::MediaSender; -use crate::photo::{self, MAX_UPLOAD_BYTES, PhotoPrep}; -use crate::queue::{PersistentTaskQueue, QueueError}; -use crate::state::EditMessage; +use input_media::{build_media_group, input_file_for, item_url}; +use post_send::{cache_animation_send, cache_sent_task}; use rand::Rng; use serde::{Deserialize, Serialize}; -use std::collections::HashMap; use std::sync::LazyLock; use teloxide::prelude::*; -use teloxide::types::{ - ChatId, InlineKeyboardButton, InlineKeyboardMarkup, InputFile, InputMedia, InputMediaAnimation, - InputMediaPhoto, InputMediaVideo, Message, MessageId, ParseMode, -}; +use teloxide::types::{ChatId, InputFile, InputMedia, MessageId}; use teloxide::{ApiError, RequestError}; -use tempfile::NamedTempFile; -use x_media::site::FetchError; +use upload::{FallbackError, PreparedItem, prepare_upload_item, send_batch_via_upload}; + +// The crate-facing API of this module lives in its submodules; re-export the +// parts other modules use so call sites stay `send::x`. +pub(crate) use post_send::{ + KEEP_ALIVE, Settled, dead_letter_notify, enqueue_retry, handle_task, post_send_actions, + settle_task, +}; /// One process-wide Bot for queue workers. Building a fresh Bot (and its HTTP /// client) per queue task was pure waste; forced at startup in main so a @@ -222,94 +230,6 @@ fn collect_file_ids(messages: &[Message], batch: &[MediaItemPayload], out: &mut } } -/// 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(ctx: &AppContext<'_>, 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) { - ctx.link_cache.put(&key, &post).await; - log::debug!("cached send for [key={}]", log_key(&post.url)); - } -} - -/// Persists a lone animation send under the post's cache key. -async fn cache_animation_send(ctx: &AppContext<'_>, task: &Task, message: &Message) { - if let Some(file_id) = message.animation().map(|a| a.file.id.to_string()) { - cache_sent_task( - ctx, - task, - vec![CachedMedia { - kind: CachedMediaKind::Animation, - file_id, - }], - ) - .await; - } -} - -/// How a task ended. The two states differ only in whether a link-cache entry -/// may still be holding the (now unusable) media. -pub enum Settled { - Sent, - Failed, -} - -/// Every path that ends a task's life — sent, permanently failed, or -/// dead-lettered after the last retry — funnels through here, so the cleanup a -/// settled task owes cannot be forgotten by a new path: release the keep-alive -/// temp media (retryable tasks keep it, they will be resent) and drop the -/// link-cache entry that a failed send's stale file ids would keep poisoning. -pub async fn settle_task(ctx: &AppContext<'_>, task: &Task, outcome: Settled) { - if matches!(outcome, Settled::Failed) { - invalidate_cache(ctx.link_cache, task).await; - } - release_keep_alive(task); -} - -/// A cached Telegram file id failed permanently (stale/expired); drop the -/// cache entry so the next request re-fetches instead of repeating it. -async fn invalidate_cache(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)); - cache.remove(&key).await; - } -} - -/// Locally produced media files (ugoira MP4, bsky remux MP4) whose temp dirs -/// must stay alive while their task may be retried by the queue. The fetch -/// pipeline hands ownership here via -/// [`x_media::site::Fetched::take_keep_alive`] before that -/// [`x_media::site::Fetched`] is dropped; a queued retry runs after that drop, -/// so without this the local file would be gone by the time the retry sends -/// it. Entries are removed when the task settles (see [`release_keep_alive`]). -pub static KEEP_ALIVE: LazyLock>> = - LazyLock::new(|| parking_lot::Mutex::new(Vec::new())); - -/// Drops the keep-alive temp dirs holding media referenced by `task` (matched -/// by path prefix). Called once a task settles — sent or permanently failed — -/// so retry-only temp files do not leak; retryable tasks keep them alive. -pub fn release_keep_alive(task: &Task) { - let paths = task.local_media_paths(); - if paths.is_empty() { - return; - } - let mut alive = KEEP_ALIVE.lock(); - alive.retain(|dir| { - let dir_path = dir.path(); - !paths.iter().any(|p| p.starts_with(dir_path)) - }); -} - pub const MAX_MEDIA_GROUP: usize = 9; /// Splits media into batches of at most [`MAX_MEDIA_GROUP`] items, moving the @@ -453,459 +373,6 @@ impl SendError { } } -fn parse_media_url(s: &str) -> Result { - url::Url::parse(s).map_err(|e| format!("invalid media URL: {e}")) -} - -fn item_url(item: &MediaItemPayload) -> &str { - match item { - MediaItemPayload::Photo { media, .. } - | MediaItemPayload::Video { media, .. } - | MediaItemPayload::Animation { media, .. } => media, - } -} - -/// Remote http(s) URLs are handed to Telegram to fetch; everything else -/// (e.g. a locally encoded ugoira MP4) is uploaded directly. -fn input_file_for(media: &str) -> Result { - if media.starts_with("http://") || media.starts_with("https://") { - Ok(InputFile::url(parse_media_url(media)?)) - } else if !std::path::Path::new(media).exists() { - // A retried task may reference a temp file the original send's - // TempDir already cleaned up; fail fast and permanent instead of - // burning retries on a file that can never come back. - Err(format!("local media file missing: {media}")) - } else { - Ok(InputFile::file(media)) - } -} - -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 { - photo = photo.caption(caption); - } - if spoiler { - photo = photo.spoiler(); - } - InputMedia::Photo(photo) -} - -fn video_media(file: InputFile, caption: Option<&str>, spoiler: bool) -> InputMedia { - let mut video = InputMediaVideo::new(file).parse_mode(ParseMode::Html); - if let Some(caption) = caption { - video = video.caption(caption); - } - if spoiler { - video = video.spoiler(); - } - InputMedia::Video(video) -} - -fn animation_media(file: InputFile, caption: Option<&str>, spoiler: bool) -> InputMedia { - let mut animation = InputMediaAnimation::new(file).parse_mode(ParseMode::Html); - if let Some(caption) = caption { - animation = animation.caption(caption); - } - if spoiler { - animation = animation.spoiler(); - } - InputMedia::Animation(animation) -} - -/// Builds a media group from payloads; only the first item of the batch gets -/// the caption (Telegram rejects captions on later items). -fn build_media_group( - batch: &[MediaItemPayload], - caption: Option<&str>, -) -> Result, String> { - batch - .iter() - .enumerate() - .map(|(i, item)| { - let item_caption = if i == 0 { caption } else { None }; - Ok(match item { - MediaItemPayload::Photo { has_spoiler, .. } => { - photo_media(item.input_file()?, item_caption, *has_spoiler) - } - MediaItemPayload::Video { - has_spoiler, - thumbnail, - .. - } => { - 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 { has_spoiler, .. } => { - animation_media(item.input_file()?, item_caption, *has_spoiler) - } - }) - }) - .collect() -} - -/// Infers a file extension from magic bytes so Telegram detects the mime type -/// on multipart uploads. -fn sniff_ext(bytes: &[u8]) -> &'static str { - if bytes.starts_with(&[0xFF, 0xD8]) { - "jpg" - } else if bytes.starts_with(b"\x89PNG") { - "png" - } else if bytes.starts_with(b"RIFF") && bytes.len() >= 12 && &bytes[8..12] == b"WEBP" { - "webp" - } else if bytes.starts_with(b"GIF8") { - "gif" - } else if bytes.len() >= 12 && &bytes[4..8] == b"ftyp" { - "mp4" - } else { - "bin" - } -} - -enum FallbackError { - Retryable { - delay_seconds: f64, - }, - Permanent { - message: String, - }, - /// The downloaded file exceeds the upload cap; the caller falls back to - /// the item's smaller URL. - MediaTooLarge, -} - -/// Brings a downloaded photo within Telegram's limits via the pure-Rust -/// chain in [`crate::photo`] (no ffmpeg): dimension cap / upload cap -/// exceeded photos are decoded, downscaled with Lanczos3, PNG bit depth -/// reduced (>24-bit → 24-bit RGB, ≤24-bit untouched) and transcoded to JPEG -/// only if still too big. Anything that cannot be fixed falls back to the -/// item's smaller URL. -/// -/// Downloads one media item to a temp file (deleted on drop), returning the -/// file plus the downloaded bytes (photos keep the bytes for -/// [`photo::prepare_photo`] — re-reading the file would double the I/O). -/// Network errors are retryable; size over the upload cap and other download -/// errors are not. -async fn download_to_temp( - item: &MediaItemPayload, -) -> Result<(NamedTempFile, bytes::Bytes), FallbackError> { - let media_url = match item { - MediaItemPayload::Photo { media, .. } - | MediaItemPayload::Video { media, .. } - | MediaItemPayload::Animation { media, .. } => media, - }; - // Photos are downloaded even over the upload cap so `prepare_photo` can - // downscale / transcode them (cap = decode budget); videos/animations - // abort as soon as the upload cap is crossed mid-stream. - let limit = if matches!(item, MediaItemPayload::Photo { .. }) { - photo::MAX_DECODE_BYTES - } else { - MAX_UPLOAD_BYTES + 1 - }; - let bytes = match x_media::site::download_media_limited(media_url, limit).await { - Ok(bytes) => bytes, - Err(FetchError::Http(_)) => { - return Err(FallbackError::Retryable { - delay_seconds: retry_delay_seconds(0), - }); - } - Err(FetchError::TooLarge) => { - return Err(FallbackError::MediaTooLarge); - } - Err(e) => { - return Err(FallbackError::Permanent { - message: format!("download failed: {e}"), - }); - } - }; - let ext = sniff_ext(&bytes); - let mut file = tempfile::Builder::new() - .suffix(&format!(".{ext}")) - .tempfile() - .map_err(|e| FallbackError::Permanent { - message: format!("temp file failed: {e}"), - })?; - use std::io::Write; - file.as_file_mut() - .write_all(&bytes) - .map_err(|e| FallbackError::Permanent { - message: format!("temp file write failed: {e}"), - })?; - Ok((file, bytes)) -} - -/// Builds the media group item from an uploaded file. -fn media_from_file( - item: &MediaItemPayload, - path: std::path::PathBuf, - caption: Option<&str>, - thumbnail: Option<&str>, -) -> Result { - let mut media = match item { - MediaItemPayload::Photo { has_spoiler, .. } => { - photo_media(InputFile::file(path), caption, *has_spoiler) - } - MediaItemPayload::Video { has_spoiler, .. } => { - video_media(InputFile::file(path), caption, *has_spoiler) - } - MediaItemPayload::Animation { has_spoiler, .. } => { - animation_media(InputFile::file(path), caption, *has_spoiler) - } - }; - if let (Some(thumb), InputMedia::Video(v)) = (thumbnail, &mut media) { - *v = v.clone().thumbnail(input_file_for(thumb)?); - } - Ok(media) -} - -/// Builds the media group item from a (smaller) URL. -fn media_from_url( - item: &MediaItemPayload, - url: &str, - caption: Option<&str>, - thumbnail: Option<&str>, -) -> Result { - let mut media = match item { - MediaItemPayload::Photo { has_spoiler, .. } => { - photo_media(input_file_for(url)?, caption, *has_spoiler) - } - MediaItemPayload::Video { has_spoiler, .. } => { - video_media(input_file_for(url)?, caption, *has_spoiler) - } - MediaItemPayload::Animation { has_spoiler, .. } => { - animation_media(input_file_for(url)?, caption, *has_spoiler) - } - }; - if let (Some(thumb), InputMedia::Video(v)) = (thumbnail, &mut media) { - *v = v.clone().thumbnail(input_file_for(thumb)?); - } - Ok(media) -} - -/// One item prepared for the upload fallback: the ready-to-send media plus -/// the temp file that must stay on disk until the group request completes. -struct PreparedItem { - /// Original position in the batch (concurrent prep completes out of order). - index: usize, - media: InputMedia, - keep_alive: Option, -} - -/// Downloads / processes one media item for the upload fallback (see -/// [`send_batch_via_upload`]). Local files are uploaded directly; oversized -/// items fall back to their smaller URL; photos are downscaled/transcoded. -async fn prepare_upload_item( - item: MediaItemPayload, - index: usize, - caption: Option<&str>, -) -> Result { - // Locally produced files (ugoira / bsky remux MP4): nothing to download - // or shrink — upload the file directly. The send is a multipart upload, - // so the only remaining failure is an upload-cap error, which is - // permanent (a video cannot be re-encoded here). - let media_url = item_url(&item); - if !media_url.starts_with("http://") && !media_url.starts_with("https://") { - let media = media_from_file( - &item, - std::path::PathBuf::from(media_url), - caption, - item.thumbnail_url(), - ) - .map_err(|message| FallbackError::Permanent { message })?; - return Ok(PreparedItem { - index, - media, - keep_alive: None, - }); - } - // Size check before downloading/uploading: over the cap, use the - // smaller URL instead of the file. Photos are exempt — they are - // downloaded and processed (downscale / PNG→JPEG) before uploading. - let too_large = match x_media::site::media_size(media_url).await { - Ok(Some(size)) => size > MAX_UPLOAD_BYTES, - _ => false, - }; - let too_large = too_large && !matches!(item, MediaItemPayload::Photo { .. }); - if too_large { - let url = item - .fallback_url() - .ok_or_else(|| FallbackError::Permanent { - message: "media too large".into(), - })?; - let media = media_from_url(&item, url, caption, item.thumbnail_url()) - .map_err(|message| FallbackError::Permanent { message })?; - return Ok(PreparedItem { - index, - media, - keep_alive: None, - }); - } - match download_to_temp(&item).await { - Ok((file, bytes)) => { - if matches!(item, MediaItemPayload::Photo { .. }) { - // Telegram rejects photos wider+taller than 10000 px combined - // (PHOTO_INVALID_DIMENSIONS): downscale the downloaded file - // before uploading; photos that cannot be brought within the - // limits degrade to the smaller URL. CPU-heavy work runs off - // the async executor thread. - let prep = tokio::task::spawn_blocking(move || photo::prepare_photo(file, &bytes)) - .await - .map_err(|e| FallbackError::Permanent { - message: format!("photo worker panicked: {e}"), - })? - .map_err(|message| FallbackError::Permanent { message })?; - match prep { - PhotoPrep::Upload(upload) => { - let path = upload.path().to_path_buf(); - let media = media_from_file(&item, path, caption, item.thumbnail_url()) - .map_err(|message| FallbackError::Permanent { message })?; - Ok(PreparedItem { - index, - media, - keep_alive: Some(upload), - }) - } - PhotoPrep::UseFallback => { - let url = item.fallback_url().ok_or_else(|| FallbackError::Permanent { - message: "photo dimensions exceed Telegram limits and no smaller variant is available" - .into(), - })?; - let media = media_from_url(&item, url, caption, item.thumbnail_url()) - .map_err(|message| FallbackError::Permanent { message })?; - Ok(PreparedItem { - index, - media, - keep_alive: None, - }) - } - } - } else { - let path = file.path().to_path_buf(); - let media = media_from_file(&item, path, caption, item.thumbnail_url()) - .map_err(|message| FallbackError::Permanent { message })?; - Ok(PreparedItem { - index, - media, - keep_alive: Some(file), - }) - } - } - Err(FallbackError::MediaTooLarge) => { - let url = item - .fallback_url() - .ok_or_else(|| FallbackError::Permanent { - message: "media too large".into(), - })?; - let media = media_from_url(&item, url, caption, item.thumbnail_url()) - .map_err(|message| FallbackError::Permanent { message })?; - Ok(PreparedItem { - index, - media, - keep_alive: None, - }) - } - Err(e) => Err(e), - } -} - -/// Download-and-reupload fallback for one media batch. Files over the upload -/// cap are not downloaded/uploaded; the item falls back to its smaller URL -/// (which Telegram fetches itself). Items are prepared concurrently (bounded) -/// because the downloads are network-bound; the batch is then uploaded in its -/// original order. Returns the fallback-error without the task attached; -/// callers wrap it with the updated task state. -async fn send_batch_via_upload( - sender: &dyn MediaSender, - chat_id: i64, - reply_to: i64, - batch: &[MediaItemPayload], - caption: Option<&str>, - task: Task, -) -> Result, SendError> { - let sem = std::sync::Arc::new(tokio::sync::Semaphore::new(3)); - let mut set = tokio::task::JoinSet::new(); - for (i, item) in batch.iter().enumerate() { - let item_caption = if i == 0 { - caption.map(str::to_string) - } else { - None - }; - let item = item.clone(); - let sem = std::sync::Arc::clone(&sem); - set.spawn(async move { - let _permit = sem.acquire().await.expect("upload semaphore closed"); - prepare_upload_item(item, i, item_caption.as_deref()).await - }); - } - let mut prepared: Vec> = (0..batch.len()).map(|_| None).collect(); - let mut keep_alive: Vec = Vec::new(); - while let Some(joined) = set.join_next().await { - let item = match joined { - Ok(Ok(item)) => item, - // Dropping the JoinSet aborts the remaining prep tasks; their - // temp files are cleaned up on drop (short-circuit like before). - Ok(Err(e)) => return Err(SendError::from_fallback(e, task.clone())), - Err(e) => { - return Err(SendError::Permanent { - message: format!("upload worker panicked: {e}"), - task: Box::new(task), - }); - } - }; - let PreparedItem { - index, - media, - keep_alive: file_opt, - } = item; - if let Some(file) = file_opt { - keep_alive.push(file); - } - prepared[index] = Some(media); - } - let items: Vec = prepared - .into_iter() - .map(|m| m.expect("every upload item was prepared")) - .collect(); - // `keep_alive` holds the temp files until the group request completes. - let result = sender - .send_media_group(ChatId(chat_id), MessageId(reply_to as i32), items) - .await; - drop(keep_alive); - match result { - Ok(messages) => Ok(messages), - Err(e) => Err(classify_to_send_error(&e, task, "upload failed")), - } -} - fn updated_sequence_task(task: &Task, batch_index: usize, sent_message_ids: Vec) -> Task { let mut updated = task.clone(); match &mut updated { @@ -1176,268 +643,13 @@ pub async fn forward_messages(ctx: &AppContext<'_>, task: &Task) -> Result<(), S } } -/// One button per template name (column layout), then the confirm button. -/// Sorted by name: the templates live in a `HashMap`, so an unsorted walk -/// would reshuffle the buttons between prompts. -pub fn build_edit_markup(templates: &HashMap) -> InlineKeyboardMarkup { - let mut names: Vec<&String> = templates.keys().collect(); - names.sort(); - let mut rows = Vec::with_capacity(names.len() + 1); - for name in names { - rows.push(vec![InlineKeyboardButton::callback( - name.clone(), - format!("template|{name}"), - )]); - } - rows.push(vec![InlineKeyboardButton::callback( - "↩️ Confirm", - "forward", - )]); - InlineKeyboardMarkup::new(rows) -} - -/// Notifies a chat about a dead-lettered task (skips when `notify_chat_id` is -/// absent). -pub async fn notify_failure( - sender: &dyn MediaSender, - chat_id: Option, - message_id: Option, - message: &str, -) { - let Some(chat_id) = chat_id else { return }; - let reply_to = message_id.map(|id| MessageId(id as i32)); - if let Err(e) = sender - .send_message(ChatId(chat_id), message.to_string(), reply_to, None) - .await - { - log::error!("failed to notify about failed task: {e}"); - } -} - -/// After a successful send: either open the edit-before-forward prompt or -/// forward to the configured channel (with retry/queue handling). -pub async fn post_send_actions(ctx: &AppContext<'_>, task: &Task, message_ids: Vec) { - let ( - chat_id, - reply_to, - source_url, - edit_before_forward, - forward_channel_id, - notify_chat_id, - notify_message_id, - ) = match task { - Task::SendMediaSequence { - chat_id, - reply_to_message_id, - source_url, - edit_before_forward, - forward_channel_id, - notify_chat_id, - notify_message_id, - .. - } - | Task::SendAnimation { - chat_id, - reply_to_message_id, - source_url, - edit_before_forward, - forward_channel_id, - notify_chat_id, - notify_message_id, - .. - } => ( - *chat_id, - *reply_to_message_id, - source_url.clone(), - *edit_before_forward, - *forward_channel_id, - *notify_chat_id, - *notify_message_id, - ), - Task::ForwardMessages { .. } => return, - }; - - if edit_before_forward { - let keyboard = build_edit_markup(&ctx.chat_store.get(chat_id).await.template); - let prompt = ctx - .sender - .send_message( - ChatId(chat_id), - "Reply to edit message.".to_string(), - Some(MessageId(reply_to as i32)), - Some(keyboard), - ) - .await; - match prompt { - Ok(prompt_id) => { - log::info!( - "edit-before-forward prompt {prompt_id} opened for {} message(s)", - message_ids.len() - ); - let source_url = source_url.clone(); - ctx.chat_store - .update(chat_id, move |data| { - data.edit_message.insert( - prompt_id, - EditMessage { - url: source_url, - chat_id, - forward_message_ids: message_ids, - template: String::new(), - created_at: unix_now(), - }, - ); - }) - .await; - } - Err(e) => log::error!("failed to send edit prompt: {e}"), - } - return; - } - - if let Some(channel_id) = forward_channel_id { - log::info!( - "forwarding {} message(s) to channel {channel_id}", - message_ids.len() - ); - let forward_task = Task::ForwardMessages { - from_chat_id: chat_id, - to_chat_id: channel_id, - message_ids, - notify_chat_id, - notify_message_id, - }; - match forward_messages(ctx, &forward_task).await { - Ok(()) => {} - Err(SendError::Retryable { - delay_seconds, - task, - }) => { - enqueue_retry(ctx.task_queue, *task, delay_seconds).await; - } - Err(SendError::Permanent { message, .. }) => { - notify_failure( - ctx.sender, - notify_chat_id, - notify_message_id, - &format!("Task failed after retries: {message}"), - ) - .await; - } - } - } -} - -/// Enqueues a task for a later attempt (retry / forward resume). When the -/// enqueue itself fails the task can never be sent again, so its keep-alive -/// temp media is released instead of leaking until process exit. -pub 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) = queue.enqueue(payload, run_after).await { - log::error!("failed to enqueue retry: {e}"); - release_keep_alive(&task); - } -} - -/// Queue entry point: parses the stored task and dispatches. -pub async fn handle_task( - ctx: &AppContext<'_>, - payload: serde_json::Value, -) -> Result<(), QueueError> { - let task: Task = match serde_json::from_value(payload.clone()) { - Ok(task) => task, - Err(e) => { - return Err(QueueError::Permanent { - message: format!("invalid task payload: {e}"), - payload, - }); - } - }; - match task { - Task::SendMediaSequence { .. } | Task::SendAnimation { .. } => { - let message_ids = match send_media_or_animation(ctx, &task).await { - Ok(ids) => ids, - Err(SendError::Retryable { - delay_seconds, - task, - }) => { - return Err(QueueError::Retryable { - delay_seconds, - payload: serde_json::to_value(task).expect("task serializes"), - }); - } - Err(SendError::Permanent { message, task }) => { - settle_task(ctx, &task, Settled::Failed).await; - return Err(QueueError::Permanent { - message, - payload: serde_json::to_value(task).expect("task serializes"), - }); - } - }; - // A task only reaches the queue after a failed send, so this - // successful run is the first time post_send_actions can fire — - // the fresh attempt failed before it ever got here. Run it - // unconditionally: `post_send_actions` executes once, after the - // whole sequence (every batch) completed, so the channel forward - // and the edit-before-forward prompt must not be lost just - // because the send needed a retry. - post_send_actions(ctx, &task, message_ids).await; - settle_task(ctx, &task, Settled::Sent).await; - Ok(()) - } - Task::ForwardMessages { .. } => match forward_messages(ctx, &task).await { - Ok(()) => Ok(()), - Err(SendError::Retryable { - delay_seconds, - task, - }) => Err(QueueError::Retryable { - delay_seconds, - payload: serde_json::to_value(task).expect("task serializes"), - }), - Err(SendError::Permanent { message, task }) => { - settle_task(ctx, &task, Settled::Failed).await; - Err(QueueError::Permanent { - message, - payload: serde_json::to_value(task).expect("task serializes"), - }) - } - }, - } -} - -async fn send_media_or_animation(ctx: &AppContext<'_>, task: &Task) -> Result, SendError> { - match task { - Task::SendMediaSequence { .. } => send_media_sequence(ctx, task).await, - Task::SendAnimation { .. } => send_animation(ctx, task).await, - Task::ForwardMessages { .. } => unreachable!(), - } -} - -/// Dead-letter callback wired to the queue in main: settles the task and -/// notifies its chat. -pub async fn dead_letter_notify(ctx: &AppContext<'_>, payload: serde_json::Value, message: String) { - // A dead-lettered task never runs again, and the queue dead-letters retry - // exhaustion itself (the handler is not called again), so this is the only - // place that sees the final payload. - if let Ok(task) = serde_json::from_value::(payload.clone()) { - settle_task(ctx, &task, Settled::Failed).await; - } - let notify_chat_id = payload.get("notify_chat_id").and_then(|v| v.as_i64()); - let notify_message_id = payload.get("notify_message_id").and_then(|v| v.as_i64()); - notify_failure( - ctx.sender, - notify_chat_id, - notify_message_id, - &format!("Task failed after retries: {message}"), - ) - .await; -} - #[cfg(test)] mod tests { + use super::post_send::build_edit_markup; + use super::upload::sniff_ext; use super::*; use crate::ctx::test_support::TestStores; + use std::collections::HashMap; use std::time::Duration; #[test] diff --git a/crates/xmedia-bot/src/send/post_send.rs b/crates/xmedia-bot/src/send/post_send.rs new file mode 100644 index 0000000..ff23051 --- /dev/null +++ b/crates/xmedia-bot/src/send/post_send.rs @@ -0,0 +1,366 @@ +//! Everything around a send: the link-cache write that follows one, the +//! keep-alive registry for locally produced media, task settlement, the +//! post-send actions (edit prompt / channel forward) and the queue entry +//! points. + +use super::{SendError, Task, forward_messages, send_animation, send_media_sequence}; +use crate::ctx::AppContext; +use crate::db::{now_f64, unix_now}; +use crate::handlers::log_key; +use crate::link_cache::{CachedMedia, CachedMediaKind, LinkCache}; +use crate::media_sender::MediaSender; +use crate::queue::{PersistentTaskQueue, QueueError}; +use crate::state::EditMessage; +use std::collections::HashMap; +use std::sync::LazyLock; +use teloxide::types::{ChatId, InlineKeyboardButton, InlineKeyboardMarkup, Message, MessageId}; + +/// 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. +pub(super) async fn cache_sent_task(ctx: &AppContext<'_>, 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) { + ctx.link_cache.put(&key, &post).await; + log::debug!("cached send for [key={}]", log_key(&post.url)); + } +} + +/// Persists a lone animation send under the post's cache key. +pub(super) async fn cache_animation_send(ctx: &AppContext<'_>, task: &Task, message: &Message) { + if let Some(file_id) = message.animation().map(|a| a.file.id.to_string()) { + cache_sent_task( + ctx, + task, + vec![CachedMedia { + kind: CachedMediaKind::Animation, + file_id, + }], + ) + .await; + } +} + +/// How a task ended. The two states differ only in whether a link-cache entry +/// may still be holding the (now unusable) media. +pub(crate) enum Settled { + Sent, + Failed, +} + +/// Every path that ends a task's life — sent, permanently failed, or +/// dead-lettered after the last retry — funnels through here, so the cleanup a +/// settled task owes cannot be forgotten by a new path: release the keep-alive +/// temp media (retryable tasks keep it, they will be resent) and drop the +/// link-cache entry that a failed send's stale file ids would keep poisoning. +pub(crate) async fn settle_task(ctx: &AppContext<'_>, task: &Task, outcome: Settled) { + if matches!(outcome, Settled::Failed) { + invalidate_cache(ctx.link_cache, task).await; + } + release_keep_alive(task); +} + +/// A cached Telegram file id failed permanently (stale/expired); drop the +/// cache entry so the next request re-fetches instead of repeating it. +async fn invalidate_cache(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)); + cache.remove(&key).await; + } +} + +/// Locally produced media files (ugoira MP4, bsky remux MP4) whose temp dirs +/// must stay alive while their task may be retried by the queue. The fetch +/// pipeline hands ownership here via +/// [`x_media::site::Fetched::take_keep_alive`] before that +/// [`x_media::site::Fetched`] is dropped; a queued retry runs after that drop, +/// so without this the local file would be gone by the time the retry sends +/// it. Entries are removed when the task settles (see [`release_keep_alive`]). +pub(crate) static KEEP_ALIVE: LazyLock>> = + LazyLock::new(|| parking_lot::Mutex::new(Vec::new())); + +/// Drops the keep-alive temp dirs holding media referenced by `task` (matched +/// by path prefix). Called once a task settles — sent or permanently failed — +/// so retry-only temp files do not leak; retryable tasks keep them alive. +pub(crate) fn release_keep_alive(task: &Task) { + let paths = task.local_media_paths(); + if paths.is_empty() { + return; + } + let mut alive = KEEP_ALIVE.lock(); + alive.retain(|dir| { + let dir_path = dir.path(); + !paths.iter().any(|p| p.starts_with(dir_path)) + }); +} + +/// One button per template name (column layout), then the confirm button. +/// Sorted by name: the templates live in a `HashMap`, so an unsorted walk +/// would reshuffle the buttons between prompts. +pub(super) fn build_edit_markup(templates: &HashMap) -> InlineKeyboardMarkup { + let mut names: Vec<&String> = templates.keys().collect(); + names.sort(); + let mut rows = Vec::with_capacity(names.len() + 1); + for name in names { + rows.push(vec![InlineKeyboardButton::callback( + name.clone(), + format!("template|{name}"), + )]); + } + rows.push(vec![InlineKeyboardButton::callback( + "↩️ Confirm", + "forward", + )]); + InlineKeyboardMarkup::new(rows) +} + +/// Notifies a chat about a dead-lettered task (skips when `notify_chat_id` is +/// absent). +pub(super) async fn notify_failure( + sender: &dyn MediaSender, + chat_id: Option, + message_id: Option, + message: &str, +) { + let Some(chat_id) = chat_id else { return }; + let reply_to = message_id.map(|id| MessageId(id as i32)); + if let Err(e) = sender + .send_message(ChatId(chat_id), message.to_string(), reply_to, None) + .await + { + log::error!("failed to notify about failed task: {e}"); + } +} + +/// After a successful send: either open the edit-before-forward prompt or +/// forward to the configured channel (with retry/queue handling). +pub(crate) async fn post_send_actions(ctx: &AppContext<'_>, task: &Task, message_ids: Vec) { + let ( + chat_id, + reply_to, + source_url, + edit_before_forward, + forward_channel_id, + notify_chat_id, + notify_message_id, + ) = match task { + Task::SendMediaSequence { + chat_id, + reply_to_message_id, + source_url, + edit_before_forward, + forward_channel_id, + notify_chat_id, + notify_message_id, + .. + } + | Task::SendAnimation { + chat_id, + reply_to_message_id, + source_url, + edit_before_forward, + forward_channel_id, + notify_chat_id, + notify_message_id, + .. + } => ( + *chat_id, + *reply_to_message_id, + source_url.clone(), + *edit_before_forward, + *forward_channel_id, + *notify_chat_id, + *notify_message_id, + ), + Task::ForwardMessages { .. } => return, + }; + + if edit_before_forward { + let keyboard = build_edit_markup(&ctx.chat_store.get(chat_id).await.template); + let prompt = ctx + .sender + .send_message( + ChatId(chat_id), + "Reply to edit message.".to_string(), + Some(MessageId(reply_to as i32)), + Some(keyboard), + ) + .await; + match prompt { + Ok(prompt_id) => { + log::info!( + "edit-before-forward prompt {prompt_id} opened for {} message(s)", + message_ids.len() + ); + let source_url = source_url.clone(); + ctx.chat_store + .update(chat_id, move |data| { + data.edit_message.insert( + prompt_id, + EditMessage { + url: source_url, + chat_id, + forward_message_ids: message_ids, + template: String::new(), + created_at: unix_now(), + }, + ); + }) + .await; + } + Err(e) => log::error!("failed to send edit prompt: {e}"), + } + return; + } + + if let Some(channel_id) = forward_channel_id { + log::info!( + "forwarding {} message(s) to channel {channel_id}", + message_ids.len() + ); + let forward_task = Task::ForwardMessages { + from_chat_id: chat_id, + to_chat_id: channel_id, + message_ids, + notify_chat_id, + notify_message_id, + }; + match forward_messages(ctx, &forward_task).await { + Ok(()) => {} + Err(SendError::Retryable { + delay_seconds, + task, + }) => { + enqueue_retry(ctx.task_queue, *task, delay_seconds).await; + } + Err(SendError::Permanent { message, .. }) => { + notify_failure( + ctx.sender, + notify_chat_id, + notify_message_id, + &format!("Task failed after retries: {message}"), + ) + .await; + } + } + } +} + +/// Enqueues a task for a later attempt (retry / forward resume). When the +/// enqueue itself fails the task can never be sent again, so its keep-alive +/// temp media is released instead of leaking until process exit. +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) = queue.enqueue(payload, run_after).await { + log::error!("failed to enqueue retry: {e}"); + release_keep_alive(&task); + } +} + +/// Queue entry point: parses the stored task and dispatches. +pub(crate) async fn handle_task( + ctx: &AppContext<'_>, + payload: serde_json::Value, +) -> Result<(), QueueError> { + let task: Task = match serde_json::from_value(payload.clone()) { + Ok(task) => task, + Err(e) => { + return Err(QueueError::Permanent { + message: format!("invalid task payload: {e}"), + payload, + }); + } + }; + match task { + Task::SendMediaSequence { .. } | Task::SendAnimation { .. } => { + let message_ids = match send_media_or_animation(ctx, &task).await { + Ok(ids) => ids, + Err(SendError::Retryable { + delay_seconds, + task, + }) => { + return Err(QueueError::Retryable { + delay_seconds, + payload: serde_json::to_value(task).expect("task serializes"), + }); + } + Err(SendError::Permanent { message, task }) => { + settle_task(ctx, &task, Settled::Failed).await; + return Err(QueueError::Permanent { + message, + payload: serde_json::to_value(task).expect("task serializes"), + }); + } + }; + // A task only reaches the queue after a failed send, so this + // successful run is the first time post_send_actions can fire — + // the fresh attempt failed before it ever got here. Run it + // unconditionally: `post_send_actions` executes once, after the + // whole sequence (every batch) completed, so the channel forward + // and the edit-before-forward prompt must not be lost just + // because the send needed a retry. + post_send_actions(ctx, &task, message_ids).await; + settle_task(ctx, &task, Settled::Sent).await; + Ok(()) + } + Task::ForwardMessages { .. } => match forward_messages(ctx, &task).await { + Ok(()) => Ok(()), + Err(SendError::Retryable { + delay_seconds, + task, + }) => Err(QueueError::Retryable { + delay_seconds, + payload: serde_json::to_value(task).expect("task serializes"), + }), + Err(SendError::Permanent { message, task }) => { + settle_task(ctx, &task, Settled::Failed).await; + Err(QueueError::Permanent { + message, + payload: serde_json::to_value(task).expect("task serializes"), + }) + } + }, + } +} + +async fn send_media_or_animation(ctx: &AppContext<'_>, task: &Task) -> Result, SendError> { + match task { + Task::SendMediaSequence { .. } => send_media_sequence(ctx, task).await, + Task::SendAnimation { .. } => send_animation(ctx, task).await, + Task::ForwardMessages { .. } => unreachable!(), + } +} + +/// Dead-letter callback wired to the queue in main: settles the task and +/// notifies its chat. +pub(crate) async fn dead_letter_notify( + ctx: &AppContext<'_>, + payload: serde_json::Value, + message: String, +) { + // A dead-lettered task never runs again, and the queue dead-letters retry + // exhaustion itself (the handler is not called again), so this is the only + // place that sees the final payload. + if let Ok(task) = serde_json::from_value::(payload.clone()) { + settle_task(ctx, &task, Settled::Failed).await; + } + let notify_chat_id = payload.get("notify_chat_id").and_then(|v| v.as_i64()); + let notify_message_id = payload.get("notify_message_id").and_then(|v| v.as_i64()); + notify_failure( + ctx.sender, + notify_chat_id, + notify_message_id, + &format!("Task failed after retries: {message}"), + ) + .await; +} diff --git a/crates/xmedia-bot/src/send/upload.rs b/crates/xmedia-bot/src/send/upload.rs new file mode 100644 index 0000000..f3680e4 --- /dev/null +++ b/crates/xmedia-bot/src/send/upload.rs @@ -0,0 +1,345 @@ +//! Download-and-reupload fallback: when Telegram cannot fetch a media URL +//! itself (hotlink protection), the bot downloads the file, shrinks photos +//! that exceed Telegram's limits and uploads the batch via multipart. + +use super::input_media::{animation_media, input_file_for, item_url, photo_media, video_media}; +use super::{MediaItemPayload, SendError, Task, classify_to_send_error, retry_delay_seconds}; +use crate::media_sender::MediaSender; +use crate::photo::{self, MAX_UPLOAD_BYTES, PhotoPrep}; +use teloxide::prelude::*; +use teloxide::types::{ChatId, InputFile, InputMedia, MessageId}; +use tempfile::NamedTempFile; +use x_media::site::FetchError; + +/// Infers a file extension from magic bytes so Telegram detects the mime type +/// on multipart uploads. +pub(super) fn sniff_ext(bytes: &[u8]) -> &'static str { + if bytes.starts_with(&[0xFF, 0xD8]) { + "jpg" + } else if bytes.starts_with(b"\x89PNG") { + "png" + } else if bytes.starts_with(b"RIFF") && bytes.len() >= 12 && &bytes[8..12] == b"WEBP" { + "webp" + } else if bytes.starts_with(b"GIF8") { + "gif" + } else if bytes.len() >= 12 && &bytes[4..8] == b"ftyp" { + "mp4" + } else { + "bin" + } +} + +pub(super) enum FallbackError { + Retryable { + delay_seconds: f64, + }, + Permanent { + message: String, + }, + /// The downloaded file exceeds the upload cap; the caller falls back to + /// the item's smaller URL. + MediaTooLarge, +} + +/// Brings a downloaded photo within Telegram's limits via the pure-Rust +/// chain in [`crate::photo`] (no ffmpeg): dimension cap / upload cap +/// exceeded photos are decoded, downscaled with Lanczos3, PNG bit depth +/// reduced (>24-bit → 24-bit RGB, ≤24-bit untouched) and transcoded to JPEG +/// only if still too big. Anything that cannot be fixed falls back to the +/// item's smaller URL. +/// +/// Downloads one media item to a temp file (deleted on drop), returning the +/// file plus the downloaded bytes (photos keep the bytes for +/// [`photo::prepare_photo`] — re-reading the file would double the I/O). +/// Network errors are retryable; size over the upload cap and other download +/// errors are not. +async fn download_to_temp( + item: &MediaItemPayload, +) -> Result<(NamedTempFile, bytes::Bytes), FallbackError> { + let media_url = match item { + MediaItemPayload::Photo { media, .. } + | MediaItemPayload::Video { media, .. } + | MediaItemPayload::Animation { media, .. } => media, + }; + // Photos are downloaded even over the upload cap so `prepare_photo` can + // downscale / transcode them (cap = decode budget); videos/animations + // abort as soon as the upload cap is crossed mid-stream. + let limit = if matches!(item, MediaItemPayload::Photo { .. }) { + photo::MAX_DECODE_BYTES + } else { + MAX_UPLOAD_BYTES + 1 + }; + let bytes = match x_media::site::download_media_limited(media_url, limit).await { + Ok(bytes) => bytes, + Err(FetchError::Http(_)) => { + return Err(FallbackError::Retryable { + delay_seconds: retry_delay_seconds(0), + }); + } + Err(FetchError::TooLarge) => { + return Err(FallbackError::MediaTooLarge); + } + Err(e) => { + return Err(FallbackError::Permanent { + message: format!("download failed: {e}"), + }); + } + }; + let ext = sniff_ext(&bytes); + let mut file = tempfile::Builder::new() + .suffix(&format!(".{ext}")) + .tempfile() + .map_err(|e| FallbackError::Permanent { + message: format!("temp file failed: {e}"), + })?; + use std::io::Write; + file.as_file_mut() + .write_all(&bytes) + .map_err(|e| FallbackError::Permanent { + message: format!("temp file write failed: {e}"), + })?; + Ok((file, bytes)) +} + +/// Builds the media group item from an uploaded file. +fn media_from_file( + item: &MediaItemPayload, + path: std::path::PathBuf, + caption: Option<&str>, + thumbnail: Option<&str>, +) -> Result { + let mut media = match item { + MediaItemPayload::Photo { has_spoiler, .. } => { + photo_media(InputFile::file(path), caption, *has_spoiler) + } + MediaItemPayload::Video { has_spoiler, .. } => { + video_media(InputFile::file(path), caption, *has_spoiler) + } + MediaItemPayload::Animation { has_spoiler, .. } => { + animation_media(InputFile::file(path), caption, *has_spoiler) + } + }; + if let (Some(thumb), InputMedia::Video(v)) = (thumbnail, &mut media) { + *v = v.clone().thumbnail(input_file_for(thumb)?); + } + Ok(media) +} + +/// Builds the media group item from a (smaller) URL. +fn media_from_url( + item: &MediaItemPayload, + url: &str, + caption: Option<&str>, + thumbnail: Option<&str>, +) -> Result { + let mut media = match item { + MediaItemPayload::Photo { has_spoiler, .. } => { + photo_media(input_file_for(url)?, caption, *has_spoiler) + } + MediaItemPayload::Video { has_spoiler, .. } => { + video_media(input_file_for(url)?, caption, *has_spoiler) + } + MediaItemPayload::Animation { has_spoiler, .. } => { + animation_media(input_file_for(url)?, caption, *has_spoiler) + } + }; + if let (Some(thumb), InputMedia::Video(v)) = (thumbnail, &mut media) { + *v = v.clone().thumbnail(input_file_for(thumb)?); + } + Ok(media) +} + +/// One item prepared for the upload fallback: the ready-to-send media plus +/// the temp file that must stay on disk until the group request completes. +pub(super) struct PreparedItem { + /// Original position in the batch (concurrent prep completes out of order). + pub(super) index: usize, + pub(super) media: InputMedia, + pub(super) keep_alive: Option, +} + +/// Downloads / processes one media item for the upload fallback (see +/// [`send_batch_via_upload`]). Local files are uploaded directly; oversized +/// items fall back to their smaller URL; photos are downscaled/transcoded. +pub(super) async fn prepare_upload_item( + item: MediaItemPayload, + index: usize, + caption: Option<&str>, +) -> Result { + // Locally produced files (ugoira / bsky remux MP4): nothing to download + // or shrink — upload the file directly. The send is a multipart upload, + // so the only remaining failure is an upload-cap error, which is + // permanent (a video cannot be re-encoded here). + let media_url = item_url(&item); + if !media_url.starts_with("http://") && !media_url.starts_with("https://") { + let media = media_from_file( + &item, + std::path::PathBuf::from(media_url), + caption, + item.thumbnail_url(), + ) + .map_err(|message| FallbackError::Permanent { message })?; + return Ok(PreparedItem { + index, + media, + keep_alive: None, + }); + } + // Size check before downloading/uploading: over the cap, use the + // smaller URL instead of the file. Photos are exempt — they are + // downloaded and processed (downscale / PNG→JPEG) before uploading. + let too_large = match x_media::site::media_size(media_url).await { + Ok(Some(size)) => size > MAX_UPLOAD_BYTES, + _ => false, + }; + let too_large = too_large && !matches!(item, MediaItemPayload::Photo { .. }); + if too_large { + let url = item + .fallback_url() + .ok_or_else(|| FallbackError::Permanent { + message: "media too large".into(), + })?; + let media = media_from_url(&item, url, caption, item.thumbnail_url()) + .map_err(|message| FallbackError::Permanent { message })?; + return Ok(PreparedItem { + index, + media, + keep_alive: None, + }); + } + match download_to_temp(&item).await { + Ok((file, bytes)) => { + if matches!(item, MediaItemPayload::Photo { .. }) { + // Telegram rejects photos wider+taller than 10000 px combined + // (PHOTO_INVALID_DIMENSIONS): downscale the downloaded file + // before uploading; photos that cannot be brought within the + // limits degrade to the smaller URL. CPU-heavy work runs off + // the async executor thread. + let prep = tokio::task::spawn_blocking(move || photo::prepare_photo(file, &bytes)) + .await + .map_err(|e| FallbackError::Permanent { + message: format!("photo worker panicked: {e}"), + })? + .map_err(|message| FallbackError::Permanent { message })?; + match prep { + PhotoPrep::Upload(upload) => { + let path = upload.path().to_path_buf(); + let media = media_from_file(&item, path, caption, item.thumbnail_url()) + .map_err(|message| FallbackError::Permanent { message })?; + Ok(PreparedItem { + index, + media, + keep_alive: Some(upload), + }) + } + PhotoPrep::UseFallback => { + let url = item.fallback_url().ok_or_else(|| FallbackError::Permanent { + message: "photo dimensions exceed Telegram limits and no smaller variant is available" + .into(), + })?; + let media = media_from_url(&item, url, caption, item.thumbnail_url()) + .map_err(|message| FallbackError::Permanent { message })?; + Ok(PreparedItem { + index, + media, + keep_alive: None, + }) + } + } + } else { + let path = file.path().to_path_buf(); + let media = media_from_file(&item, path, caption, item.thumbnail_url()) + .map_err(|message| FallbackError::Permanent { message })?; + Ok(PreparedItem { + index, + media, + keep_alive: Some(file), + }) + } + } + Err(FallbackError::MediaTooLarge) => { + let url = item + .fallback_url() + .ok_or_else(|| FallbackError::Permanent { + message: "media too large".into(), + })?; + let media = media_from_url(&item, url, caption, item.thumbnail_url()) + .map_err(|message| FallbackError::Permanent { message })?; + Ok(PreparedItem { + index, + media, + keep_alive: None, + }) + } + Err(e) => Err(e), + } +} + +/// Download-and-reupload fallback for one media batch. Files over the upload +/// cap are not downloaded/uploaded; the item falls back to its smaller URL +/// (which Telegram fetches itself). Items are prepared concurrently (bounded) +/// because the downloads are network-bound; the batch is then uploaded in its +/// original order. Returns the fallback-error without the task attached; +/// callers wrap it with the updated task state. +pub(super) async fn send_batch_via_upload( + sender: &dyn MediaSender, + chat_id: i64, + reply_to: i64, + batch: &[MediaItemPayload], + caption: Option<&str>, + task: Task, +) -> Result, SendError> { + let sem = std::sync::Arc::new(tokio::sync::Semaphore::new(3)); + let mut set = tokio::task::JoinSet::new(); + for (i, item) in batch.iter().enumerate() { + let item_caption = if i == 0 { + caption.map(str::to_string) + } else { + None + }; + let item = item.clone(); + let sem = std::sync::Arc::clone(&sem); + set.spawn(async move { + let _permit = sem.acquire().await.expect("upload semaphore closed"); + prepare_upload_item(item, i, item_caption.as_deref()).await + }); + } + let mut prepared: Vec> = (0..batch.len()).map(|_| None).collect(); + let mut keep_alive: Vec = Vec::new(); + while let Some(joined) = set.join_next().await { + let item = match joined { + Ok(Ok(item)) => item, + // Dropping the JoinSet aborts the remaining prep tasks; their + // temp files are cleaned up on drop (short-circuit like before). + Ok(Err(e)) => return Err(SendError::from_fallback(e, task.clone())), + Err(e) => { + return Err(SendError::Permanent { + message: format!("upload worker panicked: {e}"), + task: Box::new(task), + }); + } + }; + let PreparedItem { + index, + media, + keep_alive: file_opt, + } = item; + if let Some(file) = file_opt { + keep_alive.push(file); + } + prepared[index] = Some(media); + } + let items: Vec = prepared + .into_iter() + .map(|m| m.expect("every upload item was prepared")) + .collect(); + // `keep_alive` holds the temp files until the group request completes. + let result = sender + .send_media_group(ChatId(chat_id), MessageId(reply_to as i32), items) + .await; + drop(keep_alive); + match result { + Ok(messages) => Ok(messages), + Err(e) => Err(classify_to_send_error(&e, task, "upload failed")), + } +}