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

refactor(send): model a payload's media as MediaRef, not media + file_id

Browse Source 
MediaItemPayload carried one String field with two meanings and a file_id
bool beside it to say which, in all three variants; every reader re-checked
the flag (input_file had a three-arm pattern just to find the file-id case).
MediaRef::Source/FileId says it once, and the readers now match on it —
local_media_paths, item_url, input_file and the download path each shrank to
the one branch they care about.

Fixes a real failure that the flag was hiding: send_animation built its
InputFile with input_file_for(media_url), which read a cached file id as a
local path and answered "local media file missing" — permanent. So the
second request for a single-gif post always failed (the third worked, from
the degraded entry). It now uses the payload's own input_file, and
a_cached_animation_sends_by_file_id fails without that line.

Wire shape: a queued row from before this change no longer parses, and the
queue already handles that shape (handle_task dead-letters it as an invalid
payload, and dead_letter_notify still names the post and drops the stale
cache entry).
This commit is contained in:
YoursFunny
2026-09-21 18:44:26 +08:00
parent 70d4f2ec88
commit aa70b45ae6
7 changed files with 164 additions and 110 deletions
Download Patch File Download Diff File Expand all files Collapse all files
crates/xmedia-bot/src/ctx.rs
+6 -3
View File
@@ -50,7 +50,7 @@ pub static CONTEXT: LazyLock<AppContext<'static>> =
pub(crate) mod test_support {
use super::*;
use crate::link_cache::{CachedMedia, CachedMediaKind, CachedPost};
use crate::send::MediaItemPayload;
use crate::send::{MediaItemPayload, MediaRef};
use crate::state::EditMessage;
use std::sync::Arc;
use teloxide::{ApiError, RequestError};
@@ -77,10 +77,13 @@ pub(crate) mod test_support {
/// `send`'s own tests build that case directly).
pub(crate) fn photo_item(media: &str, has_spoiler: bool, file_id: bool) -> MediaItemPayload {
MediaItemPayload::Photo {
media: media.to_string(),
media: if file_id {
MediaRef::FileId(media.to_string())
} else {
MediaRef::Source(media.to_string())
},
has_spoiler,
fallback_url: None,
file_id,
}
}
crates/xmedia-bot/src/handlers/repair.rs
+6 -7
View File
@@ -190,6 +190,7 @@ mod tests {
use super::*;
use crate::ctx::test_support::{TestStores, permanent_error, photo_item};
use crate::media_sender::test_support::MockSender;
use crate::send::MediaRef;
fn queued_task(media: &str, batch_index: usize, sent: Vec<i64>) -> Task {
Task::SendMediaSequence {
@@ -281,8 +282,8 @@ mod tests {
assert_eq!(caption, "fresh caption");
assert!(
matches!(
&media_batches[0][0],
MediaItemPayload::Photo { media, .. } if media == "https://cdn/fresh.jpg"
media_batches[0][0].media_ref(),
MediaRef::Source(media) if media == "https://cdn/fresh.jpg"
),
"fresh media must replace the lost local file"
);
@@ -330,13 +331,11 @@ mod tests {
caption,
..
} => {
let media: Vec<String> = media_batches
let media: Vec<&str> = media_batches
.iter()
.flatten()
.map(|item| match item {
MediaItemPayload::Photo { media, .. }
| MediaItemPayload::Video { media, .. }
| MediaItemPayload::Animation { media, .. } => media.clone(),
.map(|item| match item.media_ref() {
MediaRef::Source(media) | MediaRef::FileId(media) => media.as_str(),
})
.collect();
assert!(!media.is_empty(), "the fresh fetch yielded no media");
crates/xmedia-bot/src/handlers/urls.rs
+46 -30
View File
@@ -5,7 +5,7 @@ use super::{log_key, reply};
use crate::ctx::AppContext;
use crate::link_cache::{CachedMedia, CachedMediaKind, CachedPost};
use crate::media_sender::MediaSender;
use crate::send::{self, Delivery, MediaItemPayload, Task};
use crate::send::{self, Delivery, MediaItemPayload, MediaRef, Task};
use crate::state::ChatData;
use std::collections::{HashMap, HashSet};
use std::future::Future;
@@ -188,24 +188,21 @@ pub(super) fn media_to_payload(media: &Media, sensitive: bool) -> MediaItemPaylo
// A gif inside a group becomes a video item; a lone gif takes the
// animation path (see url_media).
Media::Illustration { .. } => MediaItemPayload::Photo {
media: media.url().to_string(),
media: MediaRef::Source(media.url().to_string()),
has_spoiler: sensitive,
fallback_url,
file_id: false,
},
Media::Video { .. } => MediaItemPayload::Video {
media: media.url().to_string(),
media: MediaRef::Source(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(),
media: MediaRef::Source(media.url().to_string()),
has_spoiler: sensitive,
thumbnail: thumbnail_for(media),
fallback_url,
file_id: false,
},
}
}
@@ -293,11 +290,12 @@ async fn dispatch_send(
/// the URL costs Telegram a fetch (or the upload fallback a download) and saves
/// the whole source round trip, including a ugoira encode or an HLS remux.
fn cached_media_payload(media: &CachedMedia, sensitive: bool) -> MediaItemPayload {
let has_file_id = !media.file_id.is_empty();
let source = if has_file_id {
media.file_id.clone()
// The entry's file id when it has one, else its source URL (a permanently
// failed send degrades the entry and clears the id).
let source = if media.file_id.is_empty() {
MediaRef::Source(media.url.clone())
} else {
media.url.clone()
MediaRef::FileId(media.file_id.clone())
};
// A degraded item carries no smaller variant: a fresh fetch's would, but
// the item is what the source itself sent, so an oversize is handled by the
@@ -308,19 +306,16 @@ fn cached_media_payload(media: &CachedMedia, sensitive: bool) -> MediaItemPayloa
media: source,
has_spoiler: sensitive,
fallback_url,
file_id: has_file_id,
},
CachedMediaKind::Video => MediaItemPayload::Video {
media: source,
has_spoiler: sensitive,
thumbnail: None,
fallback_url,
file_id: has_file_id,
},
CachedMediaKind::Animation => MediaItemPayload::Animation {
media: source,
has_spoiler: sensitive,
file_id: has_file_id,
},
}
}
@@ -729,6 +724,34 @@ mod tests {
assert!(entry.media[0].file_id.is_empty());
}
/// A repeat request for a single-gif post re-sends the cached file id
/// instead of failing: the animation path used to hand the id to
/// `input_file_for`, which read it as a local path and answered "local
/// media file missing" (permanent), so the second request of a gif link
/// always failed and only the third — from the degraded entry — worked.
#[tokio::test]
async fn a_cached_animation_sends_by_file_id() {
let stores = TestStores::new();
let sender = MockSender::scripted(vec![Outcome::AnimationOk], permanent_error);
let ctx = stores.ctx(&sender);
let mut entry = cached_photo();
entry.media = vec![CachedMedia {
kind: CachedMediaKind::Animation,
file_id: "AgAC-gif".into(),
url: "https://p/1.gif".into(),
}];
stores.link_cache().put("twitter:1", &entry).await;
url_media(&ctx, 1, 2, "https://x.com/u/status/1", PostSend::FromChat).await;
assert_eq!(sender.calls(), vec!["send_chat_action", "send_animation"]);
assert_eq!(
sender.animation_files(),
vec!["AgAC-gif"],
"a cached gif must go out as its file id, not as an upload"
);
}
/// The two payload shapes a cached item can take, asserted directly: the
/// file id when there is one, the source URL when the entry was degraded.
#[test]
@@ -740,16 +763,14 @@ mod tests {
};
match cached_media_payload(&with_id, true) {
MediaItemPayload::Photo {
media,
media: MediaRef::FileId(media),
has_spoiler,
file_id,
..
} => {
assert_eq!(media, "AgAC");
assert!(file_id, "a cached send must go by file id");
assert_eq!(media, "AgAC", "a cached send must go by file id");
assert!(has_spoiler);
}
_ => panic!("expected a photo payload"),
other => panic!("expected a photo payload by file id, got {other:?}"),
}
let degraded = CachedMedia {
@@ -759,16 +780,14 @@ mod tests {
};
match cached_media_payload(&degraded, false) {
MediaItemPayload::Video {
media,
file_id,
media: MediaRef::Source(media),
fallback_url,
..
} => {
assert_eq!(media, "https://v/1.mp4");
assert!(!file_id, "a degraded send must go by URL");
assert_eq!(media, "https://v/1.mp4", "a degraded send goes by URL");
assert!(fallback_url.is_none());
}
_ => panic!("expected a video payload"),
other => panic!("expected a video payload by URL, got {other:?}"),
}
}
@@ -1101,17 +1120,15 @@ mod tests {
use MediaItemPayload::{Animation, Photo, Video};
let photo = || Photo {
media: "https://p/1.jpg".into(),
media: MediaRef::Source("https://p/1.jpg".into()),
has_spoiler: false,
fallback_url: None,
file_id: false,
};
let video = || Video {
media: "https://v/1.mp4".into(),
media: MediaRef::Source("https://v/1.mp4".into()),
has_spoiler: false,
thumbnail: None,
fallback_url: None,
file_id: false,
};
// Unknown before the fetch: the pipeline starts on "typing".
@@ -1124,9 +1141,8 @@ mod tests {
ActionHint::for_items(&[
video(),
Animation {
media: "https://v/2.mp4".into(),
media: MediaRef::Source("https://v/2.mp4".into()),
has_spoiler: false,
file_id: false,
}
])
.action(),
crates/xmedia-bot/src/media_sender/test_support.rs
+32 -1
View File
@@ -12,6 +12,7 @@ use parking_lot::Mutex;
pub(crate) enum Outcome {
GroupOk,
GroupErr,
AnimationOk,
AnimationErr,
CopyOk,
CopyErr,
@@ -229,11 +230,25 @@ pub(crate) struct MockSender {
answers: Mutex<Vec<Option<String>>>,
/// `(chat, message, text)` of every text rewrite, in order.
edited_texts: Mutex<Vec<(i64, i64, String)>>,
/// What each `send_animation` handed Telegram: a URL or a file id as that
/// string, an upload as `attach://<id>`.
animation_files: Mutex<Vec<String>>,
/// Builds the error every `*Err` outcome returns (RequestError is not
/// cloneable, so the factory recreates it per call).
error: Box<dyn Fn() -> RequestError + Send + Sync>,
}
/// The smallest `Message` the send paths accept, for the outcomes that must
/// report one (`send_animation` reads its id, and its media for the cache).
pub(crate) fn mock_message(id: i64) -> Message {
serde_json::from_value(serde_json::json!({
"message_id": id,
"date": 0,
"chat": { "id": 1, "type": "private" },
}))
.expect("a minimal message deserializes")
}
impl MockSender {
/// The message id a successful `send_message` reports.
pub(crate) const SENT_ID: i64 = 1;
@@ -250,6 +265,7 @@ impl MockSender {
captions: Mutex::new(Vec::new()),
answers: Mutex::new(Vec::new()),
edited_texts: Mutex::new(Vec::new()),
animation_files: Mutex::new(Vec::new()),
error: Box::new(error),
}
}
@@ -280,6 +296,11 @@ impl MockSender {
self.edited_texts.lock().clone()
}
/// What every `send_animation` handed Telegram, in order.
pub(crate) fn animation_files(&self) -> Vec<String> {
self.animation_files.lock().clone()
}
fn next(&self, kind: &'static str) -> Outcome {
self.calls.lock().push(kind);
let script = self.script.lock();
@@ -330,10 +351,20 @@ impl MediaSender for MockSender {
_reply_to: MessageId,
_caption: &'a str,
_spoiler: bool,
_file: InputFile,
file: InputFile,
) -> BoxFuture<'a, Result<Message, RequestError>> {
// Record what Telegram was handed: a URL or a file id serializes as
// that string, an upload as `attach://<id>`. Enough to tell a cached
// send (which must not re-upload) from a fresh one.
self.animation_files.lock().push(
serde_json::to_value(&file)
.ok()
.and_then(|value| value.as_str().map(str::to_string))
.unwrap_or_default(),
);
Box::pin(async move {
match self.next("send_animation") {
Outcome::AnimationOk => Ok(mock_message(MockSender::SENT_ID)),
Outcome::AnimationErr => Err(self.error()),
other => panic!("unexpected outcome {other:?} for send_animation"),
}
crates/xmedia-bot/src/send/input_media.rs
+9 -23
View File
@@ -2,16 +2,16 @@
//! URL / local path), the per-kind `InputMedia` builders and the media-group
//! assembly with its caption rule.
use super::MediaItemPayload;
use super::{MediaItemPayload, MediaRef};
use teloxide::types::{
InputFile, InputMedia, InputMediaAnimation, InputMediaPhoto, InputMediaVideo, ParseMode,
};
/// The item's media string, whether it is a URL/path or a file id — callers
/// that need the distinction match on [`MediaRef`] themselves.
pub(super) fn item_url(item: &MediaItemPayload) -> &str {
match item {
MediaItemPayload::Photo { media, .. }
| MediaItemPayload::Video { media, .. }
| MediaItemPayload::Animation { media, .. } => media,
match item.media_ref() {
MediaRef::Source(media) | MediaRef::FileId(media) => media,
}
}
@@ -35,24 +35,10 @@ impl MediaItemPayload {
/// The input for a send: a cached file id goes out as `InputFile::file_id`
/// (no fetch, no upload), URLs go to Telegram, anything else is a local
/// path (transient upload fallback).
fn input_file(&self) -> Result<InputFile, String> {
match self {
MediaItemPayload::Photo {
media,
file_id: true,
..
}
| MediaItemPayload::Video {
media,
file_id: true,
..
}
| MediaItemPayload::Animation {
media,
file_id: true,
..
} => Ok(InputFile::file_id(media.clone().into())),
_ => input_file_for(item_url(self)),
pub(super) fn input_file(&self) -> Result<InputFile, String> {
match self.media_ref() {
MediaRef::FileId(id) => Ok(InputFile::file_id(id.clone().into())),
MediaRef::Source(media) => input_file_for(media),
}
}
}
crates/xmedia-bot/src/send/mod.rs
+53 -41
View File
@@ -19,7 +19,7 @@ pub(crate) use error::classify_to_send_error;
pub use error::{
Classification, SendError, classify_request_error, is_media_fetch_failure, is_size_error,
};
use input_media::{build_media_group, input_file_for, item_url};
use input_media::{build_media_group, item_url};
use post_send::{cache_animation_send, cache_sent_task};
use serde::{Deserialize, Serialize};
use std::borrow::Cow;
@@ -41,40 +41,58 @@ pub(crate) use post_send::{
/// missing token fails fast instead of on the first task.
pub static BOT: LazyLock<Bot> = LazyLock::new(Bot::from_env);
/// Where an item's bytes come from. One `media: String` used to carry both
/// meanings with a `file_id: bool` beside it to say which — three copies of a
/// flag every reader had to re-check.
#[derive(Serialize, Deserialize, Clone, Debug, PartialEq, Eq)]
#[serde(rename_all = "snake_case")]
pub enum MediaRef {
/// A media URL Telegram fetches itself, or a local path to upload.
Source(String),
/// A Telegram file id from the link cache: sent as-is, no fetch, no upload.
FileId(String),
}
/// A queued retry persists this payload, so its wire shape is a contract with
/// the rows already on disk: `media` used to be a bare string with a
/// `file_id: bool` beside it, and a row in that older shape no longer parses —
/// the queue dead-letters it (`handle_task`'s "invalid task payload") and the
/// dead-letter path still names the post, so the one-time upgrade cost is a
/// retry that could not be resumed anyway.
#[derive(Serialize, Deserialize, Clone, Debug)]
#[serde(tag = "kind", rename_all = "snake_case")]
pub enum MediaItemPayload {
Photo {
media: String,
media: MediaRef,
has_spoiler: bool,
/// Smaller variant used when the primary media exceeds Telegram's
/// size limits.
#[serde(default)]
fallback_url: Option<String>,
/// `media` is a Telegram file id (link-cache hit), not a URL.
#[serde(default)]
file_id: bool,
},
Video {
media: String,
media: MediaRef,
has_spoiler: bool,
thumbnail: Option<String>,
#[serde(default)]
fallback_url: Option<String>,
/// `media` is a Telegram file id (link-cache hit), not a URL.
#[serde(default)]
file_id: bool,
},
Animation {
media: String,
media: MediaRef,
has_spoiler: bool,
/// `media` is a Telegram file id (link-cache hit), not a URL.
#[serde(default)]
file_id: bool,
},
}
impl MediaItemPayload {
/// The item's media reference, whatever kind of item it is.
pub(crate) fn media_ref(&self) -> &MediaRef {
match self {
MediaItemPayload::Photo { media, .. }
| MediaItemPayload::Video { media, .. }
| MediaItemPayload::Animation { media, .. } => media,
}
}
fn fallback_url(&self) -> Option<&str> {
match self {
MediaItemPayload::Photo { fallback_url, .. }
@@ -277,12 +295,8 @@ impl Task {
};
let mut out = Vec::new();
for item in items {
let is_file_id = match item {
MediaItemPayload::Photo { file_id, .. }
| MediaItemPayload::Video { file_id, .. }
| MediaItemPayload::Animation { file_id, .. } => *file_id,
};
if is_file_id {
// A file id names a Telegram-hosted copy, not a local file.
if matches!(item.media_ref(), MediaRef::FileId(_)) {
continue;
}
let media = item_url(item);
@@ -556,14 +570,15 @@ pub async fn send_animation(ctx: &AppContext<'_>, task: &Task) -> Result<Vec<i64
let text = task_text(task);
let caption = quote_long_caption(caption, &text, ctx.config.caption_quote_text_chars);
let (media_url, has_spoiler) = match animation {
MediaItemPayload::Animation {
media, has_spoiler, ..
} => (media, *has_spoiler),
MediaItemPayload::Animation { has_spoiler, .. } => (item_url(animation), *has_spoiler),
MediaItemPayload::Photo { .. } | MediaItemPayload::Video { .. } => {
unreachable!("SendAnimation carries an Animation payload")
}
};
let url_file = match input_file_for(media_url) {
// The payload knows whether its media is a URL/path or a cached file id
// (this used to go through `input_file_for`, which treated a file id as a
// local path and answered "local media file missing").
let url_file = match animation.input_file() {
Ok(file) => file,
Err(message) => {
return Err(SendError::Permanent {
@@ -719,19 +734,17 @@ mod tests {
#[test]
fn photos_first_orders_photos_before_videos() {
use MediaItemPayload::{Animation, Photo, Video};
use MediaItemPayload::{Photo, Video};
let photo = |u: &str| Photo {
media: u.into(),
media: MediaRef::Source(u.into()),
has_spoiler: false,
fallback_url: None,
file_id: false,
};
let video = |u: &str| Video {
media: u.into(),
media: MediaRef::Source(u.into()),
has_spoiler: false,
thumbnail: None,
fallback_url: None,
file_id: false,
};
let items = vec![
video("https://v/1.mp4"),
@@ -743,10 +756,8 @@ mod tests {
// All photos first (stable: p1 before p2), then all videos in order.
let kinds: Vec<&str> = ordered
.iter()
.map(|i| match i {
Photo { media, .. } => media.as_str(),
Video { media, .. } => media.as_str(),
Animation { .. } => unreachable!(),
.map(|i| match i.media_ref() {
MediaRef::Source(media) | MediaRef::FileId(media) => media.as_str(),
})
.collect();
assert_eq!(
@@ -929,9 +940,13 @@ mod tests {
#[test]
fn media_item_payload_fallback_url_serde_default() {
// Old queued payloads without the field deserialize with None.
let json =
serde_json::json!({"kind": "photo", "media": "https://a/b.jpg", "has_spoiler": false});
// `fallback_url` is optional in the wire shape: a payload written
// without it deserializes with `None`.
let json = serde_json::json!({
"kind": "photo",
"media": {"source": "https://a/b.jpg"},
"has_spoiler": false,
});
let photo: MediaItemPayload = serde_json::from_value(json).unwrap();
assert!(matches!(
photo,
@@ -1013,17 +1028,15 @@ mod tests {
caption: "cap".into(),
media_batches: vec![
vec![MediaItemPayload::Photo {
media: "https://a/b.jpg".into(),
media: MediaRef::Source("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(),
media: MediaRef::Source("https://a/v.mp4".into()),
has_spoiler: false,
thumbnail: Some("https://a/t.jpg".into()),
fallback_url: None,
file_id: false,
}],
],
batch_index: 1,
@@ -1268,9 +1281,8 @@ mod tests {
reply_to_message_id: 2,
caption: "cap".into(),
animation: MediaItemPayload::Animation {
media: file.to_string_lossy().into_owned(),
media: MediaRef::Source(file.to_string_lossy().into_owned()),
has_spoiler: false,
file_id: false,
},
source_url: "https://x.com/u/status/1".into(),
edit_before_forward: false,
crates/xmedia-bot/src/send/upload.rs
+12 -5
View File
@@ -3,7 +3,9 @@
//! that exceed Telegram's limits and uploads the batch via multipart.
use super::input_media::{input_file_for, item_url, media_from};
use super::{MediaItemPayload, SendError, Task, classify_to_send_error, retry_delay_seconds};
use super::{
MediaItemPayload, MediaRef, SendError, Task, classify_to_send_error, retry_delay_seconds,
};
use crate::media_sender::MediaSender;
use crate::photo::{self, MAX_UPLOAD_BYTES, PhotoPrep};
use std::sync::LazyLock;
@@ -68,10 +70,15 @@ pub(super) enum FallbackError {
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,
// Only a URL/path item is ever downloaded: a file id is sent as-is (see
// `MediaItemPayload::input_file`), so this path cannot see one.
let media_url = match item.media_ref() {
MediaRef::Source(media) => media,
MediaRef::FileId(id) => {
return Err(FallbackError::Permanent {
message: format!("file id reached the download path: {id}"),
});
}
};
// Photos are downloaded even over the upload cap so `prepare_photo` can
// downscale / transcode them, up to their own download cap; videos and