From 7b816d4153105a6e0b1beab228607ec9b90e8ee0 Mon Sep 17 00:00:00 2001 From: YoursFunny Date: Thu, 24 Sep 2026 22:42:47 +0800 Subject: [PATCH] fix: split long message forwards --- crates/xmedia-bot/src/handlers/callback.rs | 1 + crates/xmedia-bot/src/handlers/repair.rs | 1 + crates/xmedia-bot/src/send/mod.rs | 117 +++++++++++++++------ crates/xmedia-bot/src/send/post_send.rs | 1 + 4 files changed, 89 insertions(+), 31 deletions(-) diff --git a/crates/xmedia-bot/src/handlers/callback.rs b/crates/xmedia-bot/src/handlers/callback.rs index 41fb38c..662332d 100644 --- a/crates/xmedia-bot/src/handlers/callback.rs +++ b/crates/xmedia-bot/src/handlers/callback.rs @@ -107,6 +107,7 @@ async fn handle_callback( from_chat_id: edit.chat_id, to_chat_id: channel_id, message_ids: edit.forward_message_ids.clone(), + forward_offset: 0, notify_chat_id: Some(chat_id), notify_message_id: Some(prompt_message_id), }; diff --git a/crates/xmedia-bot/src/handlers/repair.rs b/crates/xmedia-bot/src/handlers/repair.rs index b2ffb53..3f91125 100644 --- a/crates/xmedia-bot/src/handlers/repair.rs +++ b/crates/xmedia-bot/src/handlers/repair.rs @@ -253,6 +253,7 @@ mod tests { from_chat_id: 1, to_chat_id: 2, message_ids: vec![3], + forward_offset: 0, notify_chat_id: None, notify_message_id: None, })); diff --git a/crates/xmedia-bot/src/send/mod.rs b/crates/xmedia-bot/src/send/mod.rs index a2609bd..1098ef3 100644 --- a/crates/xmedia-bot/src/send/mod.rs +++ b/crates/xmedia-bot/src/send/mod.rs @@ -150,6 +150,8 @@ pub enum Task { from_chat_id: i64, to_chat_id: i64, message_ids: Vec, + #[serde(default)] + forward_offset: usize, notify_chat_id: Option, notify_message_id: Option, }, @@ -698,46 +700,45 @@ pub(crate) async fn send_text_post( } } -/// Copies already-sent messages to the forward channel. No download fallback: -/// the files are already on Telegram's servers. pub async fn forward_messages(ctx: &AppContext<'_>, task: &Task) -> Result<(), SendError> { let Task::ForwardMessages { from_chat_id, to_chat_id, message_ids, + forward_offset, .. } = task else { unreachable!("forward_messages requires a ForwardMessages task") }; - let message_ids = message_ids - .iter() - .map(|id| MessageId(*id as i32)) - .collect::>(); - match ctx - .sender - .copy_messages( - ChatId(*to_chat_id), - ChatId(*from_chat_id), - message_ids.clone(), - ) - .await - { - Ok(_) => { - log::info!( - "copied {} message(s) from {} to {}", - message_ids.len(), - from_chat_id, - to_chat_id - ); - Ok(()) + let mut offset = (*forward_offset).min(message_ids.len()); + while offset < message_ids.len() { + let end = (offset + 100).min(message_ids.len()); + let ids = message_ids[offset..end] + .iter() + .copied() + .map(|id| MessageId(id as i32)) + .collect(); + if let Err(e) = ctx + .sender + .copy_messages(ChatId(*to_chat_id), ChatId(*from_chat_id), ids) + .await + { + let mut retry_task = task.clone(); + if let Task::ForwardMessages { forward_offset, .. } = &mut retry_task { + *forward_offset = offset; + } + return Err(classify_to_send_error(&e, retry_task, "media fetch failed")); } - Err(e) => Err(classify_to_send_error( - &e, - task.clone(), - "media fetch failed", - )), + offset = end; } + log::info!( + "copied {} message(s) from {} to {}", + message_ids.len(), + from_chat_id, + to_chat_id + ); + Ok(()) } #[cfg(test)] @@ -950,6 +951,7 @@ mod tests { from_chat_id: 1, to_chat_id: 2, message_ids: vec![1], + forward_offset: 0, notify_chat_id: None, notify_message_id: None, }; @@ -1447,6 +1449,7 @@ mod tests { from_chat_id: 1, to_chat_id: 2, message_ids: vec![3], + forward_offset: 0, notify_chat_id: None, notify_message_id: None, }; @@ -1475,11 +1478,63 @@ mod tests { )); } + #[tokio::test] + async fn long_forward_is_split_into_telegram_batches() { + let sender = MockSender::scripted(vec![Outcome::CopyOk, Outcome::CopyOk], || { + api_error("unused") + }); + let stores = TestStores::new(); + let ctx = stores.ctx(&sender); + let task = Task::ForwardMessages { + from_chat_id: 1, + to_chat_id: 2, + message_ids: (1..=201).collect(), + forward_offset: 0, + notify_chat_id: None, + notify_message_id: None, + }; + forward_messages(&ctx, &task).await.unwrap(); + assert_eq!( + sender.calls(), + vec!["copy_messages", "copy_messages", "copy_messages"] + ); + } + + #[tokio::test] + async fn failed_forward_resumes_after_completed_batches() { + use teloxide::types::Seconds; + let sender = MockSender::scripted(vec![Outcome::CopyOk, Outcome::CopyErr], || { + RequestError::RetryAfter(Seconds::from_seconds(7)) + }); + let stores = TestStores::new(); + let ctx = stores.ctx(&sender); + let task = Task::ForwardMessages { + from_chat_id: 1, + to_chat_id: 2, + message_ids: (1..=201).collect(), + forward_offset: 0, + notify_chat_id: None, + notify_message_id: None, + }; + + match forward_messages(&ctx, &task).await { + Err(SendError::Retryable { task, .. }) => { + let Task::ForwardMessages { forward_offset, .. } = *task else { + panic!("retry task is not a forward"); + }; + assert_eq!(forward_offset, 100); + } + other => panic!("expected retryable forward, got {other:?}"), + } + assert_eq!( + sender.calls(), + vec!["copy_messages", "copy_messages"], + "the first completed batch must not be replayed" + ); + } + #[tokio::test] async fn dead_letter_releases_keep_alive_temp_media() { - // A task that exhausts its retries is dead-lettered by the queue - // without the handler running again: the keep-alive temp dir the - // fetch pipeline handed over must not outlive the task. let dir = tempfile::tempdir().unwrap(); let file = dir.path().join("ugoira.mp4"); std::fs::write(&file, b"not-a-real-mp4").unwrap(); diff --git a/crates/xmedia-bot/src/send/post_send.rs b/crates/xmedia-bot/src/send/post_send.rs index 238372e..888b435 100644 --- a/crates/xmedia-bot/src/send/post_send.rs +++ b/crates/xmedia-bot/src/send/post_send.rs @@ -363,6 +363,7 @@ pub(crate) async fn post_send_actions(ctx: &AppContext<'_>, task: &Task, message from_chat_id: chat_id, to_chat_id: channel_id, message_ids, + forward_offset: 0, notify_chat_id, notify_message_id, };