diff --git a/AGENTS.md b/AGENTS.md index de4e31a..0d00236 100644 --- a/AGENTS.md +++ b/AGENTS.md @@ -45,7 +45,7 @@ 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, a `lease_token` fence: `lease_next` stamps a random token and every write-back (heartbeat, `delete`, `reschedule`, `mark_done`) is guarded by it, so a lease that expired and was re-leased cannot be written by its former holder — a lost lease stops the attempt instead; a finished row's `DELETE`/reschedule retried and a failed delete falling back to a `done` tombstone (the lease query and the sweep only look at `pending`/`in_progress`, so a task that already ran cannot be resurrected and re-run), `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; the sweep does notify the workers after it actually recovered a row, since a recovered task is due immediately while every worker may be parked on `notify` with no pending row to sleep on), `busy_timeout` on all connections | -| `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/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, and the module also carries the fixtures those tests share — the canonical cached post (`cached_photo`), the edit-before-forward prompt (`seed_prompt` with its `PROMPT_ID`/`FORWARDED_ID`) and a scripted API error (`api_error`) — so no two test modules keep their own copies | | `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 | diff --git a/crates/xmedia-bot/src/ctx.rs b/crates/xmedia-bot/src/ctx.rs index b15e6d0..d06bd9d 100644 --- a/crates/xmedia-bot/src/ctx.rs +++ b/crates/xmedia-bot/src/ctx.rs @@ -49,7 +49,66 @@ pub static CONTEXT: LazyLock> = #[cfg(test)] pub(crate) mod test_support { use super::*; + use crate::link_cache::{CachedMedia, CachedMediaKind, CachedPost}; + use crate::state::EditMessage; use std::sync::Arc; + use teloxide::{ApiError, RequestError}; + + /// The edit-before-forward prompt's message id, and the message the prompt + /// refers to (the one whose caption a reply swaps). + pub(crate) const PROMPT_ID: i64 = 7; + pub(crate) const FORWARDED_ID: i64 = 9; + + /// A Telegram API error, for the tests that script a failure. + pub(crate) fn api_error(message: &str) -> RequestError { + RequestError::Api(ApiError::Unknown(message.to_string())) + } + + /// The cached post every test that touches the link cache starts from: one + /// photo with a Telegram file id at the canonical URL (key `twitter:1`). + /// Tests that need another field mutate the returned value. + pub(crate) fn cached_photo() -> CachedPost { + CachedPost { + url: "https://x.com/u/status/1".into(), + caption: "cap".into(), + title: "t".into(), + content: "c".into(), + author: "a".into(), + author_url: "au".into(), + tags: String::new(), + sensitive: false, + media: vec![CachedMedia { + kind: CachedMediaKind::Photo, + file_id: "AgAC-file-id".into(), + }], + } + } + + /// Seeds the live prompt a post-send leaves behind in chat 1: the chat's + /// template, a bound forward channel (the prompt's "forward" button + /// branches on it) and the record for [`PROMPT_ID`] pointing at + /// [`FORWARDED_ID`]. `template` is the record's template — what a reply + /// swaps the caption through, `""` for none — and `created_at` backdates + /// the record for the expiry cases. + pub(crate) async fn seed_prompt(ctx: &AppContext<'_>, template: &str, created_at: i64) { + ctx.chat_store + .update(1, |data| { + data.forward_channel_id = Some(2); + data.template + .insert("tpl".to_string(), "[]".to_string()); + data.edit_message.insert( + PROMPT_ID, + EditMessage { + url: "https://x.com/u/status/1".into(), + chat_id: 1, + forward_message_ids: vec![FORWARDED_ID], + template: template.to_string(), + created_at, + }, + ); + }) + .await; + } pub(crate) struct TestStores { _dir: tempfile::TempDir, diff --git a/crates/xmedia-bot/src/handlers/callback.rs b/crates/xmedia-bot/src/handlers/callback.rs index aa0882b..bf75e75 100644 --- a/crates/xmedia-bot/src/handlers/callback.rs +++ b/crates/xmedia-bot/src/handlers/callback.rs @@ -204,52 +204,22 @@ async fn handle_callback( #[cfg(test)] mod tests { use super::*; - use crate::ctx::test_support::TestStores; + use crate::ctx::test_support::{PROMPT_ID, TestStores, api_error, seed_prompt}; use crate::media_sender::test_support::{MockSender, Outcome}; - use crate::state::EditMessage; - use teloxide::ApiError; - /// The edit-before-forward prompt's message id in these tests. - const PROMPT_ID: i64 = 7; - /// The message the prompt refers to (the one whose caption is swapped). - const FORWARDED_ID: i64 = 9; - - fn api_error() -> RequestError { - RequestError::Api(ApiError::Unknown("Bad Request: chat not found".into())) - } + /// The Telegram wording the mocks answer with: a chat the bot cannot reach. + const API_ERROR: &str = "Bad Request: chat not found"; fn callback_id() -> CallbackQueryId { CallbackQueryId("cb-1".to_string()) } - /// Seeds a live prompt record plus a forward channel and a template; - /// `created_at` backdates the record for the expiry cases. - async fn seed_prompt(ctx: &AppContext<'_>, created_at: i64) { - ctx.chat_store - .update(1, |data| { - data.forward_channel_id = Some(2); - data.template - .insert("tpl".to_string(), "[]".to_string()); - data.edit_message.insert( - PROMPT_ID, - EditMessage { - url: "https://x.com/u/status/1".into(), - chat_id: 1, - forward_message_ids: vec![FORWARDED_ID], - template: String::new(), - created_at, - }, - ); - }) - .await; - } - #[tokio::test] async fn template_button_swaps_the_caption_and_records_the_choice() { - let sender = MockSender::scripted(vec![Outcome::EditOk], api_error); + let sender = MockSender::scripted(vec![Outcome::EditOk], || api_error(API_ERROR)); let stores = TestStores::new(); let ctx = stores.ctx(&sender); - seed_prompt(&ctx, crate::db::unix_now()).await; + seed_prompt(&ctx, "", crate::db::unix_now()).await; handle_callback(&ctx, callback_id(), 1, PROMPT_ID, "template|tpl").await; @@ -266,10 +236,10 @@ mod tests { #[tokio::test] async fn forward_button_copies_then_clears_the_prompt() { - let sender = MockSender::scripted(vec![Outcome::CopyOk], api_error); + let sender = MockSender::scripted(vec![Outcome::CopyOk], || api_error(API_ERROR)); let stores = TestStores::new(); let ctx = stores.ctx(&sender); - seed_prompt(&ctx, crate::db::unix_now()).await; + seed_prompt(&ctx, "", crate::db::unix_now()).await; handle_callback(&ctx, callback_id(), 1, PROMPT_ID, "forward").await; @@ -288,10 +258,10 @@ mod tests { async fn skip_drops_the_prompt_without_forwarding() { // "skip" needs no forward channel and no scripted outcomes: it deletes // the prompt and drops the record, so no forward can ever happen. - let sender = MockSender::scripted(vec![], api_error); + let sender = MockSender::scripted(vec![], || api_error(API_ERROR)); let stores = TestStores::new(); let ctx = stores.ctx(&sender); - seed_prompt(&ctx, crate::db::unix_now()).await; + seed_prompt(&ctx, "", crate::db::unix_now()).await; handle_callback(&ctx, callback_id(), 1, PROMPT_ID, "skip").await; @@ -311,10 +281,10 @@ mod tests { #[tokio::test] async fn forward_without_a_channel_is_reported() { - let sender = MockSender::scripted(vec![], api_error); + let sender = MockSender::scripted(vec![], || api_error(API_ERROR)); let stores = TestStores::new(); let ctx = stores.ctx(&sender); - seed_prompt(&ctx, crate::db::unix_now()).await; + seed_prompt(&ctx, "", crate::db::unix_now()).await; ctx.chat_store .update(1, |data| data.forward_channel_id = None) .await; @@ -336,7 +306,7 @@ mod tests { }); let stores = TestStores::new(); let ctx = stores.ctx(&sender); - seed_prompt(&ctx, crate::db::unix_now()).await; + seed_prompt(&ctx, "", crate::db::unix_now()).await; handle_callback(&ctx, callback_id(), 1, PROMPT_ID, "forward").await; @@ -383,7 +353,7 @@ mod tests { #[tokio::test] async fn unknown_and_expired_prompts_answer_expired() { - let sender = MockSender::scripted(vec![], api_error); + let sender = MockSender::scripted(vec![], || api_error(API_ERROR)); let stores = TestStores::new(); let ctx = stores.ctx(&sender); @@ -393,7 +363,7 @@ mod tests { // A record past its TTL (nothing swept it yet) is dropped on use. let stale = crate::db::unix_now() - ctx.config.edit_message_ttl.as_secs() as i64 - 1; - seed_prompt(&ctx, stale).await; + seed_prompt(&ctx, "", stale).await; handle_callback(&ctx, callback_id(), 1, PROMPT_ID, "forward").await; assert_eq!( sender.answers(), diff --git a/crates/xmedia-bot/src/handlers/mod.rs b/crates/xmedia-bot/src/handlers/mod.rs index b2dcf81..bd80f7f 100644 --- a/crates/xmedia-bot/src/handlers/mod.rs +++ b/crates/xmedia-bot/src/handlers/mod.rs @@ -224,45 +224,19 @@ fn is_group(kind: &ChatKind) -> bool { #[cfg(test)] mod tests { use super::*; - use crate::ctx::test_support::TestStores; + use crate::ctx::test_support::{PROMPT_ID, TestStores, api_error, seed_prompt}; use crate::media_sender::test_support::{MockSender, Outcome}; - use crate::state::EditMessage; - use teloxide::ApiError; - const PROMPT_ID: i64 = 7; - const FORWARDED_ID: i64 = 9; - - fn api_error() -> RequestError { - RequestError::Api(ApiError::Unknown("Bad Request: message not found".into())) - } - - /// Seeds a prompt record; `template` names the chat template used for it - /// (empty = none, the caption gets the bare link). - async fn seed_prompt(ctx: &AppContext<'_>, template: &str) { - ctx.chat_store - .update(1, |data| { - data.template - .insert("tpl".to_string(), "[]".to_string()); - data.edit_message.insert( - PROMPT_ID, - EditMessage { - url: "https://x.com/u/status/1".into(), - chat_id: 1, - forward_message_ids: vec![FORWARDED_ID], - template: template.to_string(), - created_at: crate::db::unix_now(), - }, - ); - }) - .await; - } + /// The Telegram wording the mocks answer with: a message the bot cannot + /// edit (the prompt was deleted). + const API_ERROR: &str = "Bad Request: message not found"; #[tokio::test] async fn reply_to_a_prompt_swaps_the_caption_through_its_template() { - let sender = MockSender::scripted(vec![Outcome::EditOk], api_error); + let sender = MockSender::scripted(vec![Outcome::EditOk], || api_error(API_ERROR)); let stores = TestStores::new(); let ctx = stores.ctx(&sender); - seed_prompt(&ctx, "tpl").await; + seed_prompt(&ctx, "tpl", crate::db::unix_now()).await; let consumed = edit_message_handler(&ctx, 1, PROMPT_ID, "new caption").await; @@ -275,10 +249,10 @@ mod tests { #[tokio::test] async fn reply_text_and_url_are_escaped_into_the_caption() { - let sender = MockSender::scripted(vec![Outcome::EditOk], api_error); + let sender = MockSender::scripted(vec![Outcome::EditOk], || api_error(API_ERROR)); let stores = TestStores::new(); let ctx = stores.ctx(&sender); - seed_prompt(&ctx, "").await; + seed_prompt(&ctx, "", crate::db::unix_now()).await; edit_message_handler(&ctx, 1, PROMPT_ID, "").await; @@ -291,10 +265,10 @@ mod tests { #[tokio::test] async fn a_failed_caption_swap_still_consumes_the_reply() { - let sender = MockSender::scripted(vec![Outcome::EditErr], api_error); + let sender = MockSender::scripted(vec![Outcome::EditErr], || api_error(API_ERROR)); let stores = TestStores::new(); let ctx = stores.ctx(&sender); - seed_prompt(&ctx, "tpl").await; + seed_prompt(&ctx, "tpl", crate::db::unix_now()).await; // The edit failed (message deleted etc.); the reply must still be // swallowed instead of being treated as a link to fetch. @@ -304,7 +278,7 @@ mod tests { #[tokio::test] async fn reply_to_an_unrelated_message_is_not_consumed() { - let sender = MockSender::scripted(vec![], api_error); + let sender = MockSender::scripted(vec![], || api_error(API_ERROR)); let stores = TestStores::new(); let ctx = stores.ctx(&sender); diff --git a/crates/xmedia-bot/src/handlers/urls.rs b/crates/xmedia-bot/src/handlers/urls.rs index 9658670..6551504 100644 --- a/crates/xmedia-bot/src/handlers/urls.rs +++ b/crates/xmedia-bot/src/handlers/urls.rs @@ -622,8 +622,7 @@ async fn url_media_inner( #[cfg(test)] mod tests { use super::*; - use crate::ctx::test_support::TestStores; - use crate::link_cache::CachedMedia; + use crate::ctx::test_support::{TestStores, cached_photo}; use crate::media_sender::test_support::{MockSender, Outcome}; use std::time::Duration; use teloxide::{ApiError, RequestError}; @@ -634,23 +633,6 @@ mod tests { )) } - fn cached_photo_entry() -> CachedPost { - CachedPost { - url: "https://x.com/u/status/1".into(), - caption: "cap".into(), - title: "t".into(), - content: "c".into(), - author: "a".into(), - author_url: "au".into(), - tags: "".into(), - sensitive: false, - media: vec![CachedMedia { - kind: CachedMediaKind::Photo, - file_id: "file-1".into(), - }], - } - } - #[tokio::test] async fn cache_hit_sends_file_ids_and_invalidates_on_permanent_failure() { let stores = TestStores::new(); @@ -659,10 +641,7 @@ mod tests { permanent_error, ); let ctx = stores.ctx(&sender); - stores - .link_cache() - .put("twitter:1", &cached_photo_entry()) - .await; + stores.link_cache().put("twitter:1", &cached_photo()).await; url_media(&ctx, 1, 2, "https://x.com/u/status/1", PostSend::FromChat).await; @@ -699,7 +678,7 @@ mod tests { ] { let sender = MockSender::scripted(vec![Outcome::GroupOk], permanent_error); let ctx = stores.ctx(&sender); - let mut entry = cached_photo_entry(); + let mut entry = cached_photo(); entry.caption = format!("{prefix}{text}"); entry.title = String::new(); entry.content = text.into(); @@ -748,10 +727,7 @@ mod tests { let sender = MockSender::scripted(vec![Outcome::GroupOk, Outcome::MessageOk], permanent_error); let ctx = stores.ctx(&sender); - stores - .link_cache() - .put("twitter:1", &cached_photo_entry()) - .await; + stores.link_cache().put("twitter:1", &cached_photo()).await; seed_post_send_settings(&ctx).await; url_media(&ctx, 1, 2, "https://x.com/u/status/1", PostSend::FromChat).await; @@ -771,10 +747,7 @@ mod tests { // prompt (send_message) would panic with "unexpected outcome". let sender = MockSender::scripted(vec![Outcome::GroupOk], permanent_error); let ctx = stores.ctx(&sender); - stores - .link_cache() - .put("twitter:1", &cached_photo_entry()) - .await; + stores.link_cache().put("twitter:1", &cached_photo()).await; seed_post_send_settings(&ctx).await; url_media(&ctx, 1, 2, "https://x.com/u/status/1", PostSend::Suppressed).await; diff --git a/crates/xmedia-bot/src/link_cache.rs b/crates/xmedia-bot/src/link_cache.rs index fe20af7..d4541fa 100644 --- a/crates/xmedia-bot/src/link_cache.rs +++ b/crates/xmedia-bot/src/link_cache.rs @@ -180,23 +180,7 @@ impl LinkCache { #[cfg(test)] mod tests { use super::*; - - fn entry() -> CachedPost { - CachedPost { - url: "https://x.com/u/status/1".into(), - caption: "cap".into(), - title: "t".into(), - content: "c".into(), - author: "a".into(), - author_url: "au".into(), - tags: "".into(), - sensitive: true, - media: vec![CachedMedia { - kind: CachedMediaKind::Photo, - file_id: "AgAC...".into(), - }], - } - } + use crate::ctx::test_support::cached_photo; /// A payload written before the title/content split has no `content` /// field. It must still read back — the cache deletes what it cannot @@ -247,12 +231,12 @@ mod tests { let cache = LinkCache::new( crate::db::open_store(dir.path().join("c.db").to_str().unwrap()).unwrap(), ); - cache.put("twitter:1", &entry()).await; + cache.put("twitter:1", &cached_photo()).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..."); + assert_eq!(got.media[0].file_id, "AgAC-file-id"); } #[tokio::test] @@ -261,7 +245,7 @@ mod tests { let cache = LinkCache::new( crate::db::open_store(dir.path().join("c.db").to_str().unwrap()).unwrap(), ); - cache.put("twitter:1", &entry()).await; + cache.put("twitter:1", &cached_photo()).await; // Force the row into the past so a 1s TTL expires it. { let conn = rusqlite::Connection::open(dir.path().join("c.db")).unwrap(); @@ -314,8 +298,8 @@ mod tests { let cache = LinkCache::new( crate::db::open_store(dir.path().join("c.db").to_str().unwrap()).unwrap(), ); - cache.put("twitter:1", &entry()).await; - cache.put("pixiv:2", &entry()).await; + cache.put("twitter:1", &cached_photo()).await; + cache.put("pixiv:2", &cached_photo()).await; cache.remove("twitter:1").await; assert!( cache @@ -349,8 +333,8 @@ mod tests { let cache = LinkCache::new( crate::db::open_store(dir.path().join("c.db").to_str().unwrap()).unwrap(), ); - cache.put("twitter:1", &entry()).await; - cache.put("pixiv:2", &entry()).await; + cache.put("twitter:1", &cached_photo()).await; + cache.put("pixiv:2", &cached_photo()).await; // By key: only the matching row is removed. assert_eq!(cache.clear(Some("twitter:1")).await, 1); assert!( diff --git a/crates/xmedia-bot/src/send/mod.rs b/crates/xmedia-bot/src/send/mod.rs index bb14905..172ed6c 100644 --- a/crates/xmedia-bot/src/send/mod.rs +++ b/crates/xmedia-bot/src/send/mod.rs @@ -754,7 +754,7 @@ mod tests { use super::post_send::build_edit_markup; use super::upload::sniff_ext; use super::*; - use crate::ctx::test_support::TestStores; + use crate::ctx::test_support::{TestStores, cached_photo}; use std::collections::HashMap; use std::time::Duration; @@ -1561,20 +1561,7 @@ mod tests { forward_channel_id: None, notify_chat_id: None, notify_message_id: None, - cache_data: Some(CachedPost { - url: "https://x.com/u/status/1".into(), - caption: "cap".into(), - title: "t".into(), - content: "c".into(), - author: "a".into(), - author_url: "au".into(), - tags: String::new(), - sensitive: false, - media: vec![CachedMedia { - kind: CachedMediaKind::Photo, - file_id: "AgAC-file-id".into(), - }], - }), + cache_data: Some(cached_photo()), } } @@ -1584,10 +1571,7 @@ mod tests { let stores = TestStores::new(); let ctx = stores.ctx(&sender); let task = cached_sequence_task(); - stores - .link_cache() - .put("twitter:1", &cached_sequence_cache_data()) - .await; + stores.link_cache().put("twitter:1", &cached_photo()).await; settle_task(&ctx, &task, Settled::Sent).await; @@ -1607,10 +1591,7 @@ mod tests { let stores = TestStores::new(); let ctx = stores.ctx(&sender); let task = cached_sequence_task(); - stores - .link_cache() - .put("twitter:1", &cached_sequence_cache_data()) - .await; + stores.link_cache().put("twitter:1", &cached_photo()).await; settle_task(&ctx, &task, Settled::Failed).await; @@ -1623,14 +1604,4 @@ mod tests { "a permanently failed cached send must drop the entry" ); } - - fn cached_sequence_cache_data() -> CachedPost { - match cached_sequence_task() { - Task::SendMediaSequence { - cache_data: Some(post), - .. - } => post, - other => panic!("expected a cached sequence task, got {other:?}"), - } - } }