From f63af7081cc373bd77cf42246f2fd515011119c9 Mon Sep 17 00:00:00 2001 From: Valeriy Selitskiy <239034+iamwavecut@users.noreply.github.com> Date: Sun, 8 Nov 2026 00:39:35 +0100 Subject: [PATCH] refactor: untangle provider routing and durable delivery --- .env.example | 1 - Cargo.lock | 4 +- crates/openplotva-app/src/asr.rs | 40 +- crates/openplotva-app/src/callbacks.rs | 122 +- crates/openplotva-app/src/checkin.rs | 716 +- crates/openplotva-app/src/control_jobs.rs | 683 +- .../src/control_jobs/test_support.rs | 190 + crates/openplotva-app/src/delete_drawing.rs | 191 +- crates/openplotva-app/src/delete_lyrics.rs | 174 +- crates/openplotva-app/src/delete_message.rs | 140 +- crates/openplotva-app/src/dialog_debounce.rs | 8 +- .../openplotva-app/src/dialog_jobs/effects.rs | 103 +- crates/openplotva-app/src/dialog_jobs/mod.rs | 1 + crates/openplotva-app/src/dialog_messages.rs | 333 +- crates/openplotva-app/src/dialog_runtime.rs | 971 +- crates/openplotva-app/src/dialog_tools.rs | 10 +- .../openplotva-app/src/dialog_turn/session.rs | 7 +- crates/openplotva-app/src/embedder.rs | 40 +- crates/openplotva-app/src/gradius_ads.rs | 17 +- .../src/gradius_utility_outbox.rs | 26 +- crates/openplotva-app/src/help.rs | 114 +- crates/openplotva-app/src/history_summary.rs | 528 +- crates/openplotva-app/src/http/assets.rs | 36 + crates/openplotva-app/src/http/auth.rs | 304 + crates/openplotva-app/src/http/common.rs | 185 + crates/openplotva-app/src/http/entities.rs | 1260 ++ crates/openplotva-app/src/http/ingress.rs | 46 + crates/openplotva-app/src/http/memory.rs | 943 ++ crates/openplotva-app/src/http/mod.rs | 75 + .../openplotva-app/src/http/observability.rs | 1033 ++ crates/openplotva-app/src/http/redis.rs | 197 + crates/openplotva-app/src/http/routing.rs | 1183 ++ crates/openplotva-app/src/http/settings.rs | 2107 +++ crates/openplotva-app/src/http/shield.rs | 570 + crates/openplotva-app/src/http/taskman.rs | 685 + crates/openplotva-app/src/image_jobs.rs | 119 +- crates/openplotva-app/src/lib.rs | 11228 ++-------------- crates/openplotva-app/src/llm_translate.rs | 13 +- crates/openplotva-app/src/media.rs | 472 +- crates/openplotva-app/src/members.rs | 587 +- crates/openplotva-app/src/memory_runtime.rs | 833 +- crates/openplotva-app/src/memory_workers.rs | 194 + crates/openplotva-app/src/model_routing.rs | 499 +- crates/openplotva-app/src/music_jobs.rs | 805 +- .../src/openrouter_free_pool.rs | 19 +- crates/openplotva-app/src/payments.rs | 1628 +-- crates/openplotva-app/src/permissions.rs | 4 +- .../openplotva-app/src/provider_transport.rs | 400 + crates/openplotva-app/src/python_sandbox.rs | 1 + crates/openplotva-app/src/rate_limits.rs | 4 +- crates/openplotva-app/src/rates.rs | 502 +- crates/openplotva-app/src/rich.rs | 410 +- crates/openplotva-app/src/routed_attempts.rs | 38 +- crates/openplotva-app/src/runtime_cache.rs | 12 +- .../openplotva-app/src/runtime_dispatcher.rs | 507 + crates/openplotva-app/src/runtime_entities.rs | 445 +- crates/openplotva-app/src/runtime_llm.rs | 12 +- crates/openplotva-app/src/runtime_redis.rs | 94 + crates/openplotva-app/src/runtime_routing.rs | 22 + crates/openplotva-app/src/runtime_taskman.rs | 12 +- .../src/runtime_virtual_dialog.rs | 2 + crates/openplotva-app/src/settings.rs | 299 +- crates/openplotva-app/src/task_queue.rs | 180 + crates/openplotva-app/src/telegram_outbox.rs | 181 +- crates/openplotva-app/src/translate.rs | 326 +- crates/openplotva-app/src/vision.rs | 59 +- crates/openplotva-app/src/youtube.rs | 524 +- crates/openplotva-config/src/lib.rs | 12 - crates/openplotva-dialog/src/lib.rs | 3 + crates/openplotva-dialog/src/transcript.rs | 1 + crates/openplotva-llm/src/aifarm.rs | 4 +- crates/openplotva-llm/src/gemini.rs | 4 +- crates/openplotva-llm/src/gradius.rs | 6 +- crates/openplotva-llm/src/lib.rs | 1 + crates/openplotva-llm/src/provider_schema.rs | 13 +- crates/openplotva-llm/src/redaction.rs | 1191 ++ crates/openplotva-llm/src/router/table.rs | 112 + crates/openplotva-memory/Cargo.toml | 6 +- crates/openplotva-memory/src/lib.rs | 969 +- crates/openplotva-prompts/src/lib.rs | 9 +- crates/openplotva-server/Cargo.toml | 1 + crates/openplotva-server/src/lib.rs | 315 +- crates/openplotva-storage/src/app_settings.rs | 38 + crates/openplotva-storage/src/entities.rs | 423 + crates/openplotva-storage/src/gradius_ads.rs | 16 +- crates/openplotva-storage/src/lib.rs | 200 +- crates/openplotva-storage/src/llm_routing.rs | 88 +- .../openplotva-storage/src/redis_inspector.rs | 175 + crates/openplotva-storage/src/songs.rs | 260 + .../src/telegram_delivery.rs | 15 + .../openplotva-storage/src/telegram_outbox.rs | 182 +- crates/openplotva-telegram/src/lib.rs | 4 +- .../openplotva-telegram/src/update_startup.rs | 20 +- docs/CODEBASE_MAP.md | 77 +- .../213_prepared_song_delivery.down.sql | 2 + migrations/213_prepared_song_delivery.up.sql | 2 + ...4_prepared_song_delivery_identity.down.sql | 2 + ...214_prepared_song_delivery_identity.up.sql | 4 + tools/provider-smoke.sh | 27 +- 99 files changed, 18114 insertions(+), 19542 deletions(-) create mode 100644 crates/openplotva-app/src/control_jobs/test_support.rs create mode 100644 crates/openplotva-app/src/http/assets.rs create mode 100644 crates/openplotva-app/src/http/auth.rs create mode 100644 crates/openplotva-app/src/http/common.rs create mode 100644 crates/openplotva-app/src/http/entities.rs create mode 100644 crates/openplotva-app/src/http/ingress.rs create mode 100644 crates/openplotva-app/src/http/memory.rs create mode 100644 crates/openplotva-app/src/http/mod.rs create mode 100644 crates/openplotva-app/src/http/observability.rs create mode 100644 crates/openplotva-app/src/http/redis.rs create mode 100644 crates/openplotva-app/src/http/routing.rs create mode 100644 crates/openplotva-app/src/http/settings.rs create mode 100644 crates/openplotva-app/src/http/shield.rs create mode 100644 crates/openplotva-app/src/http/taskman.rs create mode 100644 crates/openplotva-app/src/memory_workers.rs create mode 100644 crates/openplotva-app/src/provider_transport.rs create mode 100644 crates/openplotva-app/src/runtime_redis.rs create mode 100644 crates/openplotva-llm/src/redaction.rs create mode 100644 crates/openplotva-storage/src/app_settings.rs create mode 100644 crates/openplotva-storage/src/entities.rs create mode 100644 crates/openplotva-storage/src/redis_inspector.rs create mode 100644 crates/openplotva-storage/src/songs.rs create mode 100644 migrations/213_prepared_song_delivery.down.sql create mode 100644 migrations/213_prepared_song_delivery.up.sql create mode 100644 migrations/214_prepared_song_delivery_identity.down.sql create mode 100644 migrations/214_prepared_song_delivery_identity.up.sql diff --git a/.env.example b/.env.example index b863f7df..9831085a 100644 --- a/.env.example +++ b/.env.example @@ -300,7 +300,6 @@ # Bootstrap/default client settings for the managed memory_extraction and # memory_subject_merge workflows; their model order and retry budgets are edited # independently in the routing admin after the first seed. -# MEMORY_CONSOLIDATION_PROVIDER=aifarm # MEMORY_CONSOLIDATION_MODEL=qwen3.8-27b # MEMORY_CONSOLIDATION_TIMEOUT_SECONDS=600 # Startup and each HH:02 UTC tick enqueue all missing completed chat hours diff --git a/Cargo.lock b/Cargo.lock index 7bf88728..a54cc5d8 100644 --- a/Cargo.lock +++ b/Cargo.lock @@ -2230,15 +2230,12 @@ dependencies = [ name = "openplotva-memory" version = "0.1.0" dependencies = [ - "base64", "futures-util", - "reqwest", "serde", "serde_json", "thiserror", "time", "tokio", - "url", ] [[package]] @@ -2299,6 +2296,7 @@ dependencies = [ "time", "tokio", "tokio-rustls", + "tower", "tracing", "x509-parser", ] diff --git a/crates/openplotva-app/src/asr.rs b/crates/openplotva-app/src/asr.rs index fc34a49c..ebf9e7cd 100644 --- a/crates/openplotva-app/src/asr.rs +++ b/crates/openplotva-app/src/asr.rs @@ -619,41 +619,19 @@ fn routed_asr_client( mut config: DiscoveryAsrConfig, attempt: &RoutedAttempt, ) -> Result { - if let Some(endpoint) = attempt - .model_base_url - .as_deref() - .or(attempt.provider_endpoint.as_deref()) - .map(str::trim) - .filter(|value| !value.is_empty()) - { + if attempt.protocol != Some(openplotva_llm::provider_schema::Protocol::DiscoveryAsr) { + return Err(AsrClientError::Discovery( + "selected protocol does not support this Discovery adapter".to_owned(), + )); + } + if let Some(endpoint) = attempt.endpoint() { config.base_url = endpoint.to_owned(); } - if let Some(service_name) = attempt - .discovery_service_name - .as_deref() - .or_else(|| { - attempt - .provider_config - .get("service_name") - .and_then(serde_json::Value::as_str) - }) - .map(str::trim) - .filter(|value| !value.is_empty()) - { + let (service_name, endpoint_name) = attempt.discovery_names(); + if let Some(service_name) = service_name { config.service_name = service_name.to_owned(); } - if let Some(endpoint_name) = attempt - .discovery_endpoint_name - .as_deref() - .or_else(|| { - attempt - .provider_config - .get("endpoint_name") - .and_then(serde_json::Value::as_str) - }) - .map(str::trim) - .filter(|value| !value.is_empty()) - { + if let Some(endpoint_name) = endpoint_name { config.endpoint_name = endpoint_name.to_owned(); } if attempt.model_name.trim().is_empty() { diff --git a/crates/openplotva-app/src/callbacks.rs b/crates/openplotva-app/src/callbacks.rs index 07fada35..7c8e3ae4 100644 --- a/crates/openplotva-app/src/callbacks.rs +++ b/crates/openplotva-app/src/callbacks.rs @@ -3,12 +3,12 @@ use std::{fmt, future::Future, pin::Pin, sync::Arc}; use carapax::types::{ - CallbackQuery as TelegramCallbackQuery, MaybeInaccessibleMessage, Update as TelegramUpdate, - UpdateType, + CallbackQuery as TelegramCallbackQuery, ChatMember as TelegramChatMember, + MaybeInaccessibleMessage, Update as TelegramUpdate, UpdateType, }; use openplotva_telegram::{ - CallbackHandlerKind, CallbackQueryRoute, TelegramOutboundMethod, callback_query_ack_method, - callback_query_route, settings_callback_ack_method, + CallbackHandlerKind, CallbackQueryRoute, TelegramOutboundMethod, build_get_chat_member_method, + callback_query_ack_method, callback_query_route, settings_callback_ack_method, }; use thiserror::Error; use time::OffsetDateTime; @@ -18,6 +18,120 @@ use crate::{ updates::UpdateHandler, }; +/// Boxed future returned by Telegram membership APIs. +pub type DeleteMemberFuture<'a, E> = + Pin> + Send + 'a>>; + +pub trait DeleteMemberApi { + /// Concrete API error. + type Error: fmt::Display + Send + Sync + 'static; + + /// Fetch one Telegram chat member. + fn get_chat_member<'a>( + &'a self, + chat_id: i64, + user_id: i64, + ) -> DeleteMemberFuture<'a, Self::Error>; +} + +impl DeleteMemberApi for openplotva_telegram::TelegramClient { + type Error = carapax::api::ExecuteError; + + fn get_chat_member<'a>( + &'a self, + chat_id: i64, + user_id: i64, + ) -> DeleteMemberFuture<'a, Self::Error> { + Box::pin(async move { + self.execute(build_get_chat_member_method(chat_id, user_id)) + .await + }) + } +} + +/// Boxed future returned by delete callback effects. +pub type DeleteEffectFuture<'a, E> = Pin> + Send + 'a>>; + +pub trait DeleteCallbackEffects { + /// Concrete effect error. + type Error: fmt::Display + Send + Sync + 'static; + + /// Execute one direct Telegram Bot API method. + fn execute_delete_method<'a>( + &'a self, + method: TelegramOutboundMethod, + ) -> DeleteEffectFuture<'a, Self::Error>; +} + +impl DeleteCallbackEffects for openplotva_telegram::TelegramClient { + type Error = carapax::api::ExecuteError; + + fn execute_delete_method<'a>( + &'a self, + method: TelegramOutboundMethod, + ) -> DeleteEffectFuture<'a, Self::Error> { + Box::pin(async move { method.execute_with(self).await.map(|_| ()) }) + } +} + +#[must_use] +pub fn telegram_member_can_delete(member: &TelegramChatMember) -> bool { + match member { + TelegramChatMember::Creator(_) => true, + TelegramChatMember::Administrator(admin) => admin.can_delete_messages, + TelegramChatMember::Kicked(_) + | TelegramChatMember::Left(_) + | TelegramChatMember::Member { .. } + | TelegramChatMember::Restricted(_) => false, + } +} + +pub(crate) async fn try_execute_delete_method( + effects: &Effects, + method: TelegramOutboundMethod, + context: &'static str, +) where + Effects: DeleteCallbackEffects + Sync, +{ + let method_name = method.method_name(); + if let Err(error) = effects.execute_delete_method(method).await { + tracing::warn!( + message = %error, + method = method_name, + context, + "generic delete callback Telegram side effect failed" + ); + } +} + +pub(crate) async fn can_delete_owned_content( + member_api: &MemberApi, + query: &TelegramCallbackQuery, + data_user_id: i64, + data_chat_id: i64, +) -> bool +where + MemberApi: DeleteMemberApi + Sync, +{ + let requester_id = i64::from(query.from.id); + if requester_id == data_user_id { + return true; + } + + match member_api.get_chat_member(data_chat_id, requester_id).await { + Ok(member) => telegram_member_can_delete(&member), + Err(error) => { + tracing::warn!( + message = %error, + chat_id = data_chat_id, + user_id = requester_id, + "failed to authorize delete callback" + ); + false + } + } +} + /// Boxed future returned by callback-query side effects. pub type CallbackQueryFuture<'a, E> = Pin> + Send + 'a>>; diff --git a/crates/openplotva-app/src/checkin.rs b/crates/openplotva-app/src/checkin.rs index 0105d476..aad467f2 100644 --- a/crates/openplotva-app/src/checkin.rs +++ b/crates/openplotva-app/src/checkin.rs @@ -1,5 +1,8 @@ //! App-level check-in callback control-job behavior. +#[cfg(test)] +use crate::control_jobs::ControlJobEnqueueFuture; +use crate::control_jobs::ControlJobQueue; use std::{ collections::{HashMap, HashSet}, fmt, @@ -15,8 +18,8 @@ use carapax::types::{ User as TelegramUser, }; use openplotva_taskman::{ - CONTROL_QUEUE_NAME, ControlJobData, ControlJobParams, ControlKind, HIGH_PRIORITY, JobType, - StatelessJobItem, control_job_params_from_stateless_job, new_control_job_at, + CONTROL_QUEUE_NAME, ControlJobData, ControlJobParams, ControlKind, HIGH_PRIORITY, + StatelessJobItem, new_control_job_at, }; use openplotva_telegram::{ ChatRef, DispatcherConfig, DispatcherQueue, InlineKeyboardMarkup, OUTBOUND_SEND_MAX_ATTEMPTS, @@ -55,16 +58,14 @@ const CHECKIN_EPHEMERAL_TIMEOUT: Duration = Duration::from_secs(30); const NEW_CHAT_WARMUP_WINDOW: time::Duration = time::Duration::hours(24); const CHECKIN_DIRECT_RETRY_AFTER_INLINE_CAP_SECS: u64 = 15; -/// Boxed future returned by check-in control-job queue calls. -pub type CheckinControlJobQueueFuture<'a, E> = - Pin> + Send + 'a>>; - -/// Boxed future returned by check-in control-job worker queue calls. -pub type CheckinControlJobWorkerFuture<'a, T, E> = - Pin> + Send + 'a>>; - /// Boxed future returned by check-in control-job effects. -pub type CheckinControlJobFuture<'a, E> = Pin> + Send + 'a>>; +pub type CheckinControlJobFuture<'a, E> = Pin< + Box< + dyn Future, E>> + + Send + + 'a, + >, +>; /// Boxed future returned by check-in game storage calls. pub type CheckinGameStoreFuture<'a, T, E> = Pin> + Send + 'a>>; @@ -87,17 +88,6 @@ pub type CheckinThemeEffectFuture<'a, E> = Pin = Pin> + Send + 'a>>; -pub trait CheckinControlJobQueue { - /// Error returned by the concrete queue implementation. - type Error: fmt::Display + Send + Sync + 'static; - - fn assign_checkin_control_job<'a>( - &'a self, - queue_name: &'static str, - job: StatelessJobItem, - ) -> CheckinControlJobQueueFuture<'a, Self::Error>; -} - pub trait CheckinCommandPermission { /// Error returned by the concrete permission policy. type Error: fmt::Display + Send + Sync + 'static; @@ -136,92 +126,13 @@ where } } -impl CheckinControlJobQueue for T -where - T: crate::payments::PaymentControlJobQueue + Sync, -{ - type Error = T::Error; - - fn assign_checkin_control_job<'a>( - &'a self, - queue_name: &'static str, - job: StatelessJobItem, - ) -> CheckinControlJobQueueFuture<'a, Self::Error> { - self.assign_payment_control_job(queue_name, job) - } -} - -/// Queue/status boundary for check-in taskman control-job workers. -pub trait CheckinControlJobWorkerQueue { - /// Queue error type. - type Error: fmt::Display + Send + Sync + 'static; - - fn dequeue_checkin_control_job<'a>( - &'a self, - queue_name: &'static str, - ) -> CheckinControlJobWorkerFuture< - 'a, - Option, - Self::Error, - >; - - /// Mark one check-in control job completed. - fn complete_checkin_control_job<'a>( - &'a self, - job_id: i64, - ) -> CheckinControlJobWorkerFuture<'a, (), Self::Error>; - - /// Mark one check-in control job failed. - fn fail_checkin_control_job<'a>( - &'a self, - job_id: i64, - error: &'a str, - ) -> CheckinControlJobWorkerFuture<'a, (), Self::Error>; -} - -impl CheckinControlJobWorkerQueue for T -where - T: crate::payments::SharedControlJobWorkerQueue + Sync, -{ - type Error = T::Error; - - fn dequeue_checkin_control_job<'a>( - &'a self, - queue_name: &'static str, - ) -> CheckinControlJobWorkerFuture< - 'a, - Option, - Self::Error, - > { - self.dequeue_shared_control_job_matching( - queue_name, - "checkin-control", - is_checkin_control_job, - ) - } - - fn complete_checkin_control_job<'a>( - &'a self, - job_id: i64, - ) -> CheckinControlJobWorkerFuture<'a, (), Self::Error> { - self.complete_shared_control_job(job_id) - } - - fn fail_checkin_control_job<'a>( - &'a self, - job_id: i64, - error: &'a str, - ) -> CheckinControlJobWorkerFuture<'a, (), Self::Error> { - self.fail_shared_control_job(job_id, error) - } -} - pub trait CheckinControlJobEffects { /// Error returned by the concrete game runner. type Error: fmt::Display + Send + Sync + 'static; fn run_checkin_game<'a>( &'a self, + job_id: i64, params: &'a ControlJobParams, ) -> CheckinControlJobFuture<'a, Self::Error>; } @@ -318,6 +229,7 @@ pub struct CheckinGameMessage { #[derive(Clone, Debug, Eq, PartialEq)] pub struct CheckinGameRunRequest { + pub job_id: i64, /// Reconstructed control-job message context. pub message: CheckinGameMessage, /// Telegram chat type string. @@ -343,6 +255,19 @@ pub trait CheckinGameSender { /// Error returned by the concrete sender. type Error: fmt::Display + Send + Sync + 'static; + fn queued_checkin<'a>( + &'a self, + job_id: i64, + ) -> CheckinGameSendFuture<'a, Option, Self::Error>; + + fn finish_checkin_html<'a>( + &'a self, + job_id: i64, + reply_to: CheckinGameMessage, + target_message_id: Option, + html: String, + ) -> CheckinGameSendFuture<'a, crate::telegram_outbox::QueuedBatchReceipt, Self::Error>; + /// Send one immediate HTML reply. Returns Telegram message ID when available. fn send_checkin_html<'a>( &'a self, @@ -517,31 +442,6 @@ impl CheckinGameStore for PostgresCheckinGameStore { } } -#[derive(Clone, Debug, Eq, PartialEq)] -pub enum CheckinGameRunOutcome { - /// Private chat or text-send permission blocked the game before side effects. - SkippedByChatPolicy, - /// Game disabled in chat settings. - Disabled, - /// Existing daily winner was reported. - ExistingWinner, - /// Not enough valid participants. - NotEnoughParticipants { - /// Valid participant count. - count: usize, - }, - /// New winner recorded, stats increment attempted, and winner message sent. - WinnerRecorded { - /// Winner user ID. - user_id: i64, - /// Theme key used for storage/output. - theme: String, - }, - /// Insert raced with another worker; existing winner was reported when visible. - RecordRaceExistingWinner, - RecordFailedNoWinner, -} - /// Recoverable check-in game runner errors. #[derive(Clone, Debug, Eq, Error, PartialEq)] pub enum CheckinGameRunError { @@ -570,6 +470,7 @@ pub struct CheckinGameRuntimeEffects { #[derive(Clone)] pub struct TelegramCheckinGameSender { + delivery: Arc, ephemeral_store: openplotva_storage::RedisEphemeralMessageStore, telegram: openplotva_telegram::TelegramClient, rich: openplotva_telegram::RichApiClient, @@ -583,8 +484,10 @@ impl TelegramCheckinGameSender { ephemeral_store: openplotva_storage::RedisEphemeralMessageStore, telegram: openplotva_telegram::TelegramClient, rich: openplotva_telegram::RichApiClient, + delivery: Arc, ) -> Self { Self { + delivery, ephemeral_store, telegram, rich, @@ -603,6 +506,59 @@ impl TelegramCheckinGameSender { impl CheckinGameSender for TelegramCheckinGameSender { type Error = CheckinGameSenderError; + fn queued_checkin<'a>( + &'a self, + job_id: i64, + ) -> CheckinGameSendFuture<'a, Option, Self::Error> + { + Box::pin(async move { + self.delivery + .queued(&checkin_delivery_source(job_id), Some(job_id)) + .await + .map_err(|error| CheckinGameSenderError::Telegram(error.to_string())) + }) + } + + fn finish_checkin_html<'a>( + &'a self, + job_id: i64, + reply_to: CheckinGameMessage, + target_message_id: Option, + html: String, + ) -> CheckinGameSendFuture<'a, crate::telegram_outbox::QueuedBatchReceipt, Self::Error> { + Box::pin(async move { + let method = match target_message_id { + Some(message_id) => { + TelegramOutboundMethod::from(openplotva_telegram::EditRichMessage { + chat_id: reply_to.chat_id, + message_id: i64::from(message_id), + html, + reply_markup: None, + }) + } + None => TelegramOutboundMethod::from(openplotva_telegram::SendRichMessage { + chat_id: reply_to.chat_id, + html, + options: openplotva_telegram::RichSendOptions { + message_thread_id: reply_to.thread_id.map(i64::from), + reply_to_message_id: Some(i64::from(reply_to.message_id)), + allow_sending_without_reply: true, + ..Default::default() + }, + }), + }; + self.delivery + .enqueue( + &checkin_delivery_source(job_id), + reply_to.thread_id.map(i64::from), + method, + Some(job_id), + ) + .await + .map_err(|error| CheckinGameSenderError::Telegram(error.to_string())) + }) + } + fn send_checkin_html<'a>( &'a self, reply_to: CheckinGameMessage, @@ -673,19 +629,19 @@ pub enum CheckinGameSenderError { } #[derive(Clone)] -pub struct CheckinCommandDispatcherEffects { +pub struct CheckinCommandDelivery { queue: Arc, - rich: Arc, + delivery: Arc, next_virtual_id: VirtualIdFactory, } -impl CheckinCommandDispatcherEffects { +impl CheckinCommandDelivery { /// Build check-in command effects over an existing dispatcher queue. #[must_use] - pub fn new(queue: Arc, rich: Arc) -> Self { + pub fn new(queue: Arc, delivery: Arc) -> Self { Self { queue, - rich, + delivery, next_virtual_id: monotonic_virtual_id_factory("checkin-command-vmsg"), } } @@ -707,7 +663,7 @@ pub enum CheckinCommandEffectError { RichSend(String), } -impl CheckinCommandEffects for CheckinCommandDispatcherEffects { +impl CheckinCommandEffects for CheckinCommandDelivery { type Error = CheckinCommandEffectError; fn send_checkin_today_winner_with_stats<'a>( @@ -724,8 +680,20 @@ impl CheckinCommandEffects for CheckinCommandDispatcherEffects { disable_notification: false, reply_markup: None, }; - self.rich - .send_rich(reply_to.chat_id, &html, &options) + self.delivery + .enqueue( + &format!( + "checkin-command:{}:{}", + reply_to.chat_id, reply_to.message_id + ), + options.message_thread_id, + openplotva_telegram::SendRichMessage { + chat_id: reply_to.chat_id, + html, + options, + }, + None, + ) .await .map_err(|error| CheckinCommandEffectError::RichSend(error.to_string()))?; Ok(()) @@ -1018,9 +986,18 @@ where fn run_checkin_game<'a>( &'a self, + job_id: i64, params: &'a ControlJobParams, ) -> CheckinControlJobFuture<'a, Self::Error> { Box::pin(async move { + if let Some(receipt) = self + .sender + .queued_checkin(job_id) + .await + .map_err(sender_error)? + { + return Ok(Some(receipt)); + } let message = CheckinGameMessage { chat_id: params.chat_id, message_id: params.message_id, @@ -1041,6 +1018,7 @@ where &self.store, &self.sender, CheckinGameRunRequest { + job_id, message, chat_type: params.data.chat_type.clone(), theme_override: params.data.theme.clone(), @@ -1060,17 +1038,17 @@ where }), ) .await - .map(|_| ()) }) } } -pub async fn run_checkin_game_at( +#[cfg(test)] +async fn run_checkin_game_at( store: &Store, sender: &Sender, request: CheckinGameRunRequest, pick_index: Pick, -) -> Result +) -> Result, CheckinGameRunError> where Store: CheckinGameStore + Sync, Sender: CheckinGameSender + Sync, @@ -1093,14 +1071,14 @@ async fn run_checkin_game_inner( pick_index: &Pick, sleep_between_steps: bool, utility_ads: Option>, -) -> Result +) -> Result, CheckinGameRunError> where Store: CheckinGameStore + Sync, Sender: CheckinGameSender + Sync, Pick: Fn(usize) -> usize + Sync + ?Sized, { if request.chat_type.eq_ignore_ascii_case("private") || !request.can_send_text { - return Ok(CheckinGameRunOutcome::SkippedByChatPolicy); + return Ok(None); } let message = request.message; @@ -1117,12 +1095,13 @@ where ) .await .map_err(sender_error)?; - return Ok(CheckinGameRunOutcome::Disabled); + return Ok(None); } if let Some(winner) = load_today_winner(store, message.chat_id).await { - send_today_winner_message(store, sender, message, &winner).await; - return Ok(CheckinGameRunOutcome::ExistingWinner); + let receipt = + send_today_winner_message(store, sender, request.job_id, message, &winner).await?; + return Ok(Some(receipt)); } let theme = resolve_checkin_theme(&settings, &request.theme_override); @@ -1139,9 +1118,7 @@ where ) .await .map_err(sender_error)?; - return Ok(CheckinGameRunOutcome::NotEnoughParticipants { - count: participants.len(), - }); + return Ok(None); } let picked = pick_index(participants.len()).min(participants.len() - 1); @@ -1157,7 +1134,7 @@ where { tracing::warn!(message = %error, chat_id = message.chat_id, winner_id, "failed to increment check-in game stats"); } - run_checkin_animation( + let receipt = run_checkin_animation( store, sender, message, @@ -1166,21 +1143,20 @@ where pick_index, sleep_between_steps, utility_ads, - request.now, + request.job_id, ) .await?; - Ok(CheckinGameRunOutcome::WinnerRecorded { - user_id: winner_id, - theme: theme.key.to_owned(), - }) + Ok(Some(receipt)) } Err(error) => { tracing::warn!(message = %error, chat_id = message.chat_id, winner_id, "failed to record check-in daily winner"); if let Some(existing) = load_today_winner(store, message.chat_id).await { - send_today_winner_message(store, sender, message, &existing).await; - Ok(CheckinGameRunOutcome::RecordRaceExistingWinner) + let receipt = + send_today_winner_message(store, sender, request.job_id, message, &existing) + .await?; + Ok(Some(receipt)) } else { - Ok(CheckinGameRunOutcome::RecordFailedNoWinner) + Ok(None) } } } @@ -1333,6 +1309,10 @@ fn daily_game_disabled(settings: &openplotva_core::ChatSettings) -> bool { settings.enable_daily_game == Some(false) } +pub(crate) fn checkin_delivery_source(job_id: i64) -> String { + format!("checkin-final:job:{job_id}") +} + #[allow(clippy::too_many_arguments)] async fn run_checkin_animation( store: &Store, @@ -1343,8 +1323,8 @@ async fn run_checkin_animation( pick_index: &Pick, sleep_between_steps: bool, utility_ads: Option>, - game_at: OffsetDateTime, -) -> Result<(), CheckinGameRunError> + job_id: i64, +) -> Result where Store: CheckinGameStore + Sync, Sender: CheckinGameSender + Sync, @@ -1362,7 +1342,8 @@ where .ok() .flatten() else { - return Ok(()); + let linked_name = winner_link(store, winner_id).await; + return sender.finish_checkin_html(job_id, message, None, theme.winner_text(&linked_name)).await.map_err(sender_error); }; for step in progress.iter().skip(1) { if sleep_between_steps { @@ -1386,17 +1367,7 @@ where text.push_str("\n\n"); text.push_str(&theme.winner_text(&linked_name)); if let Some(utility) = utility_ads { - if let Err(_error) = sender - .edit_checkin_html(message, message_id, text.clone()) - .await - { - tracing::warn!( - chat_id = message.chat_id, - message_id, - "failed to edit check-in winner message before optional ad" - ); - } - let source_id = format!("{}:{}", message.chat_id, game_at.date()); + let source_id = format!("job:{job_id}"); let result_context = openplotva_telegram::strip_telegram_html(&text); let prepared = utility .ads @@ -1404,10 +1375,7 @@ where surface: crate::gradius_ads::GradiusUtilitySurface::Checkin, include_vip_appendix: false, source_id: source_id.clone(), - attempt_key: format!( - "checkin-final:{source_id}:{}", - OffsetDateTime::now_utc().unix_timestamp_nanos() - ), + attempt_key: checkin_delivery_source(job_id), user_id: utility.initiator_id, chat_id: message.chat_id, thread_id: message.thread_id, @@ -1430,10 +1398,19 @@ where message.thread_id, i64::from(message_id), html, + job_id, ) .await { - Ok(_) => return Ok(()), + Ok(_) => { + return sender + .queued_checkin(job_id) + .await + .map_err(sender_error)? + .ok_or_else(|| CheckinGameRunError::Sender { + message: "check-in final batch was not persisted".into(), + }); + } Err(_error) => { tracing::warn!( opportunity_id = ad.opportunity_id, @@ -1453,17 +1430,11 @@ where ) } } - return Ok(()); } - if let Err(error) = sender.edit_checkin_html(message, message_id, text).await { - tracing::warn!( - message = %error, - chat_id = message.chat_id, - message_id, - "failed to edit check-in winner message" - ); - } - Ok(()) + sender + .finish_checkin_html(job_id, message, Some(message_id), text) + .await + .map_err(sender_error) } fn checkin_progress_lines(theme: CheckinGameTheme, pick_index: &Pick) -> Vec @@ -1492,9 +1463,11 @@ where async fn send_today_winner_message( store: &Store, sender: &Sender, + job_id: i64, message: CheckinGameMessage, winner: &openplotva_storage::ChatGameResult, -) where +) -> Result +where Store: CheckinGameStore + Sync, Sender: CheckinGameSender + Sync, { @@ -1505,9 +1478,10 @@ async fn send_today_winner_message( theme.name, theme.winner_text(&linked_name) ); - if let Err(error) = sender.send_checkin_html(message, text).await { - tracing::warn!(message = %error, chat_id = message.chat_id, "failed to send today's check-in winner"); - } + sender + .finish_checkin_html(job_id, message, None, text) + .await + .map_err(sender_error) } async fn today_winner_with_stats_html( @@ -2220,12 +2194,6 @@ pub enum CheckinThemeCallbackOutcome { QueueError, } -/// Result of one check-in control-job execution. -#[derive(Clone, Debug, Eq, PartialEq)] -pub enum CheckinControlJobOutcome { - Completed, -} - #[derive(Clone, Debug, Eq, PartialEq)] pub enum CheckinCommandOutcome { /// Update did not belong to this slice and was delegated. @@ -2243,28 +2211,6 @@ pub enum CheckinCommandOutcome { MissingMessage, } -/// Result of one check-in taskman worker tick. -#[derive(Clone, Debug, Default, Eq, PartialEq)] -pub struct CheckinControlJobWorkerReport { - /// Whether a job was dequeued. - pub dequeued: bool, - /// Dequeued taskman job ID. - pub job_id: Option, - /// Check-in executor result, when payload decoding succeeded. - pub execution: Option, - /// The job was finalized as completed. - pub completed: bool, - /// The job was finalized as failed. - pub failed: bool, - /// Queue dequeue failed. - pub dequeue_error: Option, - pub decode_error: Option, - /// Game runner failed. - pub execution_error: Option, - /// Completion/failure status write failed. - pub status_error: Option, -} - /// Fatal errors from the check-in callback wrapper. #[derive(Clone, Debug, Eq, Error, PartialEq)] pub enum CheckinThemeCallbackError { @@ -2303,7 +2249,7 @@ impl CheckinThemeCallbackUpdateHandler UpdateHandler for CheckinThemeCallbackUpdateHandler where - Queue: CheckinControlJobQueue + Send + Sync, + Queue: ControlJobQueue + Send + Sync, Effects: CheckinThemeEffects + Send + Sync, Next: UpdateHandler + Send + Sync, { @@ -2359,7 +2305,7 @@ impl impl UpdateHandler for CheckinCommandUpdateHandler where - Queue: CheckinControlJobQueue + Send + Sync, + Queue: ControlJobQueue + Send + Sync, Store: CheckinGameStore + Send + Sync, Permission: CheckinCommandPermission + Send + Sync, Effects: CheckinCommandEffects + Send + Sync, @@ -2417,7 +2363,7 @@ pub async fn handle_checkin_theme_callback_update_or_else_at< handle_other: HandleFn, ) -> Result where - Queue: CheckinControlJobQueue + Sync, + Queue: ControlJobQueue + Sync, Effects: CheckinThemeEffects + Sync, HandleFn: FnOnce(TelegramUpdate) -> HandleFuture, HandleFuture: Future>, @@ -2452,7 +2398,7 @@ pub async fn handle_checkin_theme_callback_at( created: OffsetDateTime, ) -> CheckinThemeCallbackOutcome where - Queue: CheckinControlJobQueue + Sync, + Queue: ControlJobQueue + Sync, Effects: CheckinThemeEffects + Sync, { let (ack, blocked) = openplotva_telegram::checkin_theme_selection_ack_method( @@ -2472,10 +2418,7 @@ where return CheckinThemeCallbackOutcome::MissingMessage; }; - if let Err(error) = queue - .assign_checkin_control_job(CONTROL_QUEUE_NAME, job) - .await - { + if let Err(error) = queue.assign_control_job(CONTROL_QUEUE_NAME, job).await { tracing::warn!(message = %error, "failed to assign check-in control job"); try_send_checkin_queue_error_notice(effects, message).await; return CheckinThemeCallbackOutcome::QueueError; @@ -2558,7 +2501,7 @@ pub async fn handle_checkin_command_update_or_else_at< handle_other: HandleFn, ) -> Result where - Queue: CheckinControlJobQueue + Sync, + Queue: ControlJobQueue + Sync, Store: CheckinGameStore + Sync, Effects: CheckinCommandEffects + Sync, HandleFn: FnOnce(TelegramUpdate) -> HandleFuture, @@ -2603,7 +2546,7 @@ pub async fn handle_checkin_command_message_at( created: OffsetDateTime, ) -> CheckinCommandOutcome where - Queue: CheckinControlJobQueue + Sync, + Queue: ControlJobQueue + Sync, Store: CheckinGameStore + Sync, Effects: CheckinCommandEffects + Sync, { @@ -2644,10 +2587,7 @@ where let Some(job) = checkin_control_job_from_message_at(message, "", created) else { return CheckinCommandOutcome::MissingMessage; }; - if let Err(error) = queue - .assign_checkin_control_job(CONTROL_QUEUE_NAME, job) - .await - { + if let Err(error) = queue.assign_control_job(CONTROL_QUEUE_NAME, job).await { tracing::warn!(message = %error, chat_id = reply_to.chat_id, "failed to assign /checkin control job"); try_send_checkin_command_queue_error(effects, reply_to).await; return CheckinCommandOutcome::QueueError; @@ -2656,89 +2596,6 @@ where CheckinCommandOutcome::ControlJobQueued } -/// Process one check-in control job, if available. -pub async fn process_checkin_control_job_once_at( - queue: &Queue, - effects: &Effects, -) -> CheckinControlJobWorkerReport -where - Queue: CheckinControlJobWorkerQueue + Sync, - Effects: CheckinControlJobEffects + Sync, -{ - let mut report = CheckinControlJobWorkerReport::default(); - let item = match queue.dequeue_checkin_control_job(CONTROL_QUEUE_NAME).await { - Ok(item) => item, - Err(error) => { - report.dequeue_error = Some(error.to_string()); - return report; - } - }; - - let Some(item) = item else { - return report; - }; - report.dequeued = true; - report.job_id = Some(item.id); - - let params = match control_job_params_from_stateless_job(&item.job) { - Ok(params) => params, - Err(error) => { - let error = error.to_string(); - report.decode_error = Some(error.clone()); - mark_checkin_control_job_failed(queue, item.id, &error, &mut report).await; - return report; - } - }; - - if params.data.kind != ControlKind::Checkin { - let error = format!( - "unsupported check-in control job kind: {:?}", - params.data.kind - ); - report.execution_error = Some(error.clone()); - mark_checkin_control_job_failed(queue, item.id, &error, &mut report).await; - return report; - } - - match effects.run_checkin_game(¶ms).await { - Ok(()) => match queue.complete_checkin_control_job(item.id).await { - Ok(()) => { - report.completed = true; - report.execution = Some(CheckinControlJobOutcome::Completed); - } - Err(error) => report.status_error = Some(error.to_string()), - }, - Err(error) => { - let error = error.to_string(); - report.execution_error = Some(error.clone()); - mark_checkin_control_job_failed(queue, item.id, &error, &mut report).await; - } - } - report -} - -async fn mark_checkin_control_job_failed( - queue: &Queue, - job_id: i64, - error: &str, - report: &mut CheckinControlJobWorkerReport, -) where - Queue: CheckinControlJobWorkerQueue + Sync, -{ - match queue.fail_checkin_control_job(job_id, error).await { - Ok(()) => report.failed = true, - Err(error) => report.status_error = Some(error.to_string()), - } -} - -fn is_checkin_control_job(job: &StatelessJobItem) -> bool { - job.data.job_type == JobType::Control - && matches!( - job.data.control_data.as_ref().map(|data| data.kind), - Some(ControlKind::Checkin) - ) -} - fn checkin_theme_callback_data( query: &TelegramCallbackQuery, ) -> Option { @@ -2905,12 +2762,11 @@ mod tests { CallbackQueryEffects, CallbackQueryFuture, CallbackQueryUpdateRoute, CallbackRateLimitFuture, CallbackRateLimitPolicy, handle_callback_query_update_or_else, }; - use crate::payments::{InMemoryPaymentControlJobQueue, InMemoryPaymentControlJobStatus}; use crate::updates::{ TelegramFileMetadataStoreFuture, UpdateHandler, UpdateStateStore, UpdateStateStoreFuture, process_update_with_state_store_at, }; - use openplotva_taskman::{JobPayload, JobType}; + use openplotva_taskman::{JobPayload, JobType, control_job_params_from_stateless_job}; use openplotva_telegram::{ CallbackHandlerKind, DispatcherConfig, DispatcherQueue, TelegramOutboundMethodKind, }; @@ -2923,6 +2779,65 @@ mod tests { time::{Duration, SystemTime, UNIX_EPOCH}, }; + #[tokio::test] + async fn live_checkin_stats_replay_reuses_the_prepared_rich_message() + -> Result<(), Box> { + let Ok(dsn) = std::env::var("OPENPLOTVA_TEST_POSTGRES_DSN") else { + return Ok(()); + }; + let pool = sqlx::postgres::PgPoolOptions::new() + .max_connections(2) + .connect(&dsn) + .await?; + openplotva_storage::run_migrations_on(&pool).await?; + let bot_id = OffsetDateTime::now_utc().unix_timestamp_nanos() as i64; + let effects = CheckinCommandDelivery::new( + Arc::new(DispatcherQueue::new(Default::default())), + Arc::new(crate::rich::RichOutbox::new( + openplotva_storage::PostgresTelegramOutboxStore::new(pool.clone()), + bot_id, + )), + ); + let reply = CheckinGameMessage { + chat_id: -10042, + message_id: 77, + thread_id: Some(9), + }; + effects + .send_checkin_today_winner_with_stats(reply, "

Winner

Ada

".into()) + .await?; + effects + .send_checkin_today_winner_with_stats(reply, "changed stats on replay".into()) + .await?; + let rows: Vec<(String, serde_json::Value)> = + sqlx::query_as("SELECT state, payload FROM telegram_outbox WHERE bot_id = $1") + .bind(bot_id) + .fetch_all(&pool) + .await?; + assert_eq!(rows.len(), 1); + assert_eq!(rows[0].0, "pending"); + assert_eq!(rows[0].1["chat_id"], -10042); + assert_eq!(rows[0].1["options"]["message_thread_id"], 9); + assert_eq!(rows[0].1["options"]["reply_to_message_id"], 77); + assert!( + rows[0].1["html"] + .as_str() + .expect("rich HTML") + .contains("Ada") + ); + assert!( + !rows[0].1["html"] + .as_str() + .expect("rich HTML") + .contains("changed stats") + ); + sqlx::query("DELETE FROM telegram_outbox WHERE bot_id = $1") + .bind(bot_id) + .execute(&pool) + .await?; + Ok(()) + } + #[tokio::test] async fn checkin_theme_callback_acks_and_queues_go_control_job() -> Result<(), Box> { @@ -3380,34 +3295,51 @@ mod tests { } #[tokio::test] - async fn checkin_control_worker_skips_other_jobs_and_runs_checkin_game() + async fn postgres_checkin_final_edit_is_durable_and_reuses_its_payload() -> Result<(), Box> { - let queue = InMemoryPaymentControlJobQueue::new(); - let effects = CheckinControlEffectsStub::default(); - let translate_job = checkin_test_control_job(ControlKind::Translate, "translate", ""); - let checkin_job = checkin_test_control_job(ControlKind::Checkin, "checkin", "classic"); - queue - .assign_checkin_control_job(CONTROL_QUEUE_NAME, translate_job) - .await?; - queue - .assign_checkin_control_job(CONTROL_QUEUE_NAME, checkin_job.clone()) + let Ok(dsn) = std::env::var("OPENPLOTVA_TEST_POSTGRES_DSN") else { + return Ok(()); + }; + let pool = sqlx::postgres::PgPoolOptions::new() + .max_connections(4) + .connect(&dsn) .await?; - - let report = process_checkin_control_job_once_at(&queue, &effects).await; - - assert!(report.dequeued); - assert_eq!(report.job_id, Some(2)); - assert_eq!(report.execution, Some(CheckinControlJobOutcome::Completed)); - assert!(report.completed); - assert!(!report.failed); - assert_eq!(effects.runs(), vec![(-10042, 777000, "classic".to_owned())]); - let snapshot = queue.snapshot(); - assert_eq!(snapshot[0].status, InMemoryPaymentControlJobStatus::Pending); - assert_eq!( - snapshot[1].status, - InMemoryPaymentControlJobStatus::Completed + openplotva_storage::run_migrations_on(&pool).await?; + let delivery = Arc::new(crate::rich::RichOutbox::new( + openplotva_storage::PostgresTelegramOutboxStore::new(pool.clone()), + 7, + )); + let sender = TelegramCheckinGameSender::new( + openplotva_storage::RedisEphemeralMessageStore::new(redis::Client::open( + "redis://127.0.0.1:1/", + )?), + openplotva_telegram::telegram_client_with_base_url("123:test", "http://127.0.0.1:1")?, + openplotva_telegram::RichApiClient::with_base_url("123:test", "http://127.0.0.1:1")?, + delivery, ); - assert_eq!(snapshot[1].job, checkin_job); + let job_id = OffsetDateTime::now_utc().unix_timestamp_nanos() as i64; + let message = game_message(); + assert!(sender.queued_checkin(job_id).await?.is_none()); + let first = sender + .finish_checkin_html(job_id, message, Some(9001), "

Winner

".into()) + .await?; + assert!(!first.delivery_complete()); + let replay = sender + .finish_checkin_html(job_id, message, Some(9001), "

Changed

".into()) + .await?; + assert_eq!(first, replay); + let saved: (serde_json::Value, String, String, Option) = sqlx::query_as( + "SELECT payload, delivery_policy, ordering_key, dialog_job_id FROM telegram_outbox WHERE batch_id = $1" + ).bind(&first.batch_id).fetch_one(&pool).await?; + assert_eq!(saved.0["message_id"], 9001); + assert_eq!(saved.0["html"], "

Winner

"); + assert_eq!(saved.1, "target_idempotent"); + assert_eq!(saved.2, "dialog:7:-10042:77"); + assert_eq!(saved.3, Some(job_id)); + sqlx::query("DELETE FROM telegram_outbox WHERE batch_id = $1") + .bind(&first.batch_id) + .execute(&pool) + .await?; Ok(()) } @@ -3432,13 +3364,7 @@ mod tests { run_checkin_game_at(&store, &sender, game_request("supergroup", "kotik"), |_| 1) .await?; - assert_eq!( - outcome, - CheckinGameRunOutcome::WinnerRecorded { - user_id: 102, - theme: "kotik".to_owned(), - } - ); + assert!(outcome.is_some()); assert_eq!(store.recorded(), vec![(-10042, 102, "kotik".to_owned())]); assert_eq!(store.incremented(), vec![(-10042, 102)]); assert_eq!(sender.ephemeral(), Vec::::new()); @@ -3506,7 +3432,7 @@ mod tests { let outcome = run_checkin_game_at(&store, &sender, game_request("group", ""), |_| 0).await?; - assert_eq!(outcome, CheckinGameRunOutcome::ExistingWinner); + assert!(outcome.is_some()); assert!(store.recorded().is_empty()); assert_eq!(sender.edits(), Vec::::new()); assert_eq!(sender.ephemeral(), Vec::::new()); @@ -3581,7 +3507,7 @@ mod tests { let outcome = run_checkin_game_at(&store, &sender, game_request("supergroup", ""), |_| 0).await?; - assert_eq!(outcome, CheckinGameRunOutcome::Disabled); + assert!(outcome.is_none()); assert_eq!( sender.ephemeral(), vec!["🚫 Π˜Π³Ρ€Π° Π²Ρ‹ΠΊΠ»ΡŽΡ‡Π΅Π½Π° Π² настройках".to_owned()] @@ -3943,30 +3869,6 @@ mod tests { }) } - fn checkin_test_control_job(kind: ControlKind, title: &str, theme: &str) -> StatelessJobItem { - new_control_job_at( - ControlJobParams { - chat_id: -10042, - message_id: 555, - user_id: 777000, - user_full_name: "Plotva Bot".to_owned(), - thread_id: Some(77), - data: ControlJobData { - kind, - theme: theme.to_owned(), - chat_type: "supergroup".to_owned(), - user_name: "plotva_bot".to_owned(), - first_name: "Plotva".to_owned(), - last_name: "Bot".to_owned(), - ..ControlJobData::default() - }, - }, - OffsetDateTime::UNIX_EPOCH, - ) - .with_name(title) - .with_priority(HIGH_PRIORITY) - } - #[derive(Default)] struct QueueStub { state: Mutex, @@ -3997,14 +3899,14 @@ mod tests { } } - impl CheckinControlJobQueue for QueueStub { + impl ControlJobQueue for QueueStub { type Error = io::Error; - fn assign_checkin_control_job<'a>( + fn assign_control_job<'a>( &'a self, queue_name: &'static str, job: StatelessJobItem, - ) -> CheckinControlJobQueueFuture<'a, Self::Error> { + ) -> ControlJobEnqueueFuture<'a, Self::Error> { Box::pin(async move { let mut state = self.lock(); state.assigned.push((queue_name, job)); @@ -4136,35 +4038,6 @@ mod tests { } } - #[derive(Default)] - struct CheckinControlEffectsStub { - runs: Mutex>, - } - - impl CheckinControlEffectsStub { - fn runs(&self) -> Vec<(i64, i64, String)> { - self.runs.lock().expect("checkin runs").clone() - } - } - - impl CheckinControlJobEffects for CheckinControlEffectsStub { - type Error = io::Error; - - fn run_checkin_game<'a>( - &'a self, - params: &'a ControlJobParams, - ) -> CheckinControlJobFuture<'a, Self::Error> { - Box::pin(async move { - self.runs.lock().expect("checkin runs").push(( - params.chat_id, - params.user_id, - params.data.theme.clone(), - )); - Ok(()) - }) - } - } - #[derive(Default)] struct GameStoreStub { state: Mutex, @@ -4372,6 +4245,38 @@ mod tests { impl CheckinGameSender for GameSenderStub { type Error = io::Error; + fn queued_checkin<'a>( + &'a self, + _job_id: i64, + ) -> CheckinGameSendFuture< + 'a, + Option, + Self::Error, + > { + Box::pin(async { Ok(None) }) + } + fn finish_checkin_html<'a>( + &'a self, + job_id: i64, + _reply_to: CheckinGameMessage, + target_message_id: Option, + html: String, + ) -> CheckinGameSendFuture<'a, crate::telegram_outbox::QueuedBatchReceipt, Self::Error> + { + Box::pin(async move { + if target_message_id.is_some() { + self.edits.lock().expect("edits").push(html); + } else { + self.sent_html.lock().expect("sent html").push(html); + } + Ok(crate::telegram_outbox::QueuedBatchReceipt::durable_outbox( + checkin_delivery_source(job_id), + vec!["checkin-final".into()], + false, + )) + }) + } + fn send_checkin_html<'a>( &'a self, _reply_to: CheckinGameMessage, @@ -4418,6 +4323,7 @@ mod tests { fn game_request(chat_type: &str, theme_override: &str) -> CheckinGameRunRequest { CheckinGameRunRequest { + job_id: 555, message: game_message(), chat_type: chat_type.to_owned(), theme_override: theme_override.to_owned(), diff --git a/crates/openplotva-app/src/control_jobs.rs b/crates/openplotva-app/src/control_jobs.rs index 5fa54f66..1a60df42 100644 --- a/crates/openplotva-app/src/control_jobs.rs +++ b/crates/openplotva-app/src/control_jobs.rs @@ -1,3 +1,545 @@ +use crate::rate_limits::TaskEnqueueRateLimit; +use openplotva_taskman::{InMemoryTaskQueue, TaskQueueRecord, TaskQueueSchedule}; +#[cfg(test)] +use serde::{Deserialize, Serialize}; +use std::sync::Arc; +use thiserror::Error; + +/// Boxed future returned by taskman control-job assignment calls. +pub type ControlJobEnqueueFuture<'a, E> = Pin> + Send + 'a>>; +/// Queue operations used by the shared control worker. +pub type ControlJobQueueFuture<'a, T, E> = Pin> + Send + 'a>>; +/// A taskman row ready for the shared control worker. +#[derive(Clone, Debug, Eq, PartialEq)] +pub struct ControlJobWorkItem { + /// Taskman job ID used for completion/failure writes. + pub id: i64, + pub job: StatelessJobItem, +} + +pub trait ControlJobQueue { + /// Error returned by the concrete queue implementation. + type Error: fmt::Display + Send + Sync + 'static; + + /// Assign a control job to a named taskman queue. + fn assign_control_job<'a>( + &'a self, + queue_name: &'static str, + job: StatelessJobItem, + ) -> ControlJobEnqueueFuture<'a, Self::Error>; + + /// Assign or replace a pending control job under a stable debounce key. + fn schedule_control_job<'a>( + &'a self, + queue_name: &'static str, + job: StatelessJobItem, + schedule: TaskQueueSchedule, + ) -> ControlJobEnqueueFuture<'a, Self::Error> { + drop(schedule); + self.assign_control_job(queue_name, job) + } +} + +pub trait SharedControlJobWorkerQueue { + /// Error returned by the concrete queue implementation. + type Error: fmt::Display + Send + Sync + 'static; + + /// Dequeue the next pending control job owned by one executor family. + fn dequeue_shared_control_job_matching<'a>( + &'a self, + queue_name: &'static str, + worker_id: &'static str, + predicate: fn(&StatelessJobItem) -> bool, + ) -> ControlJobQueueFuture<'a, Option, Self::Error>; + + /// Mark one shared control job completed. + fn complete_shared_control_job<'a>( + &'a self, + job_id: i64, + ) -> ControlJobQueueFuture<'a, (), Self::Error>; + + fn wait_for_shared_control_delivery<'a>( + &'a self, + job_id: i64, + ) -> ControlJobQueueFuture<'a, (), Self::Error>; + + /// Mark one shared control job failed. + fn fail_shared_control_job<'a>( + &'a self, + job_id: i64, + error: &'a str, + ) -> ControlJobQueueFuture<'a, (), Self::Error>; +} + +#[cfg(test)] +#[derive(Clone, Debug)] +pub struct InMemoryControlJobQueue { + queue: InMemoryTaskQueue, +} + +#[cfg(test)] +/// Queue status projection for fixtures. +#[derive(Clone, Copy, Debug, Deserialize, Eq, PartialEq, Serialize)] +pub enum InMemoryControlJobStatus { + /// Job is pending worker execution. + Pending, + /// Job has been dequeued by a worker. + Processing, + /// Job completed. + Completed, + /// Job failed and carries an error string. + Failed, +} + +#[cfg(test)] +/// Queue snapshot used by fixtures. +#[derive(Clone, Debug, Deserialize, Eq, PartialEq, Serialize)] +pub struct InMemoryControlJobRecord { + /// Taskman job ID. + pub id: i64, + pub queue_name: String, + /// Current in-memory status. + pub status: InMemoryControlJobStatus, + pub job: StatelessJobItem, + /// Worker failure. + pub error: Option, +} + +/// Rejected taskman operation. +#[derive(Clone, Debug, Error, Eq, PartialEq)] +pub enum InMemoryControlJobQueueError { + /// A status update targeted a job ID this queue has never assigned. + #[error("payment control job {0} not found")] + JobNotFound(i64), + /// The shared taskman queue core rejected an operation unexpectedly. + #[error("payment control job queue: {0}")] + Taskman(String), +} + +#[cfg(test)] +impl InMemoryControlJobQueue { + /// Build an empty control queue for fixtures. + #[must_use] + pub fn new() -> Self { + Self { + queue: InMemoryTaskQueue::new(), + } + } + + /// Return a stable ID-ordered snapshot of the queue state. + #[cfg(test)] + #[must_use] + pub fn snapshot(&self) -> Vec { + let records = self.queue.records(); + control_records_from_taskman(&records) + } + + /// Return raw taskman records with worker IDs and timestamps for runtime diagnostics. + #[must_use] + pub fn taskman_records(&self) -> Vec { + self.queue.records() + } +} + +/// Control queue storage or enqueue failure. +#[derive(Debug, Error)] +pub enum PersistentControlJobQueueError { + #[error(transparent)] + Storage(#[from] openplotva_storage::StorageError), + /// The inner queue operation failed. + #[error(transparent)] + Queue(#[from] InMemoryControlJobQueueError), + /// User/chat-originated control job creation was rejected by the task enqueue limiter. + #[error("task enqueue rate limited")] + TaskEnqueueRateLimited, +} + +/// Shared control queue with enqueue limiting and a Postgres journal. +#[derive(Clone)] +pub struct PersistentControlJobQueue { + queue: InMemoryTaskQueue, + task_enqueue_rate_limit: Option>, + db_journal: Option< + crate::task_queue::BufferedTaskQueueJournal, + >, +} + +impl PersistentControlJobQueue { + #[must_use] + pub fn from_task_queue(queue: InMemoryTaskQueue) -> Self { + Self { + queue, + task_enqueue_rate_limit: None, + db_journal: None, + } + } + + #[must_use] + pub fn with_task_enqueue_rate_limit( + mut self, + rate_limit: Arc, + ) -> Self { + self.task_enqueue_rate_limit = Some(rate_limit); + self + } + + /// Attach the shared Postgres journal so control/payment enqueues are flushed + /// to Postgres synchronously. Payment jobs (e.g. SuccessfulPayment) must be + /// durable before the update is acked: a lost enqueue would silently drop money. + #[must_use] + pub fn with_db_journal( + mut self, + journal: crate::task_queue::BufferedTaskQueueJournal< + openplotva_storage::PostgresTaskQueueStore, + >, + ) -> Self { + self.db_journal = Some(journal); + self + } + + /// Return a stable ID-ordered snapshot of the queue state. + #[cfg(test)] + #[must_use] + pub fn snapshot(&self) -> Vec { + control_records_from_taskman(&self.queue.records()) + } + + /// Return raw taskman records with worker IDs and timestamps for runtime diagnostics. + #[must_use] + pub fn taskman_records(&self) -> Vec { + self.queue.records() + } +} + +/// Synchronous-flush retry budget for payment-flow control jobs before falling back +/// to the background sync worker. +const PAYMENT_CONTROL_FLUSH_ATTEMPTS: u32 = 3; +const PAYMENT_CONTROL_FLUSH_RETRY_DELAY: std::time::Duration = std::time::Duration::from_millis(50); + +/// Control kinds that must never be dropped by enqueue rate limiting. +const fn control_kind_skips_task_enqueue_rate_limit(kind: ControlKind) -> bool { + matches!( + kind, + ControlKind::SuccessfulPayment + | ControlKind::VipInvoice + | ControlKind::DonateInvoice + | ControlKind::ChatAdminsSync + | ControlKind::ChatMemberSync + ) +} + +impl PersistentControlJobQueue { + async fn assign_control_job_with_schedule( + &self, + queue_name: &'static str, + job: StatelessJobItem, + schedule: Option, + ) -> Result<(), PersistentControlJobQueueError> { + let params = control_job_params_from_stateless_job(&job) + .map_err(|error| InMemoryControlJobQueueError::Taskman(error.to_string()))?; + let chat_id = params.chat_id; + let user_id = params.user_id; + let control_kind = params.data.kind; + // Payment and member-state control jobs are exempt from enqueue throttling: + // dropping them would silently lose durable payment intent or Telegram state. + // They are rare control-plane events, not the dialog burst the limiter targets. + let rate_limit = if control_kind_skips_task_enqueue_rate_limit(control_kind) { + None + } else { + self.task_enqueue_rate_limit.as_ref() + }; + if let Some(rate_limit) = rate_limit { + let now = OffsetDateTime::now_utc(); + let report = rate_limit + .check_task_enqueue_rate_limit(chat_id, user_id, now) + .await; + if report.limited { + tracing::warn!( + queue_name, + chat_id, + user_id, + control_kind = ?control_kind, + rate_limit_scope = ?report.scope, + rate_limit_error = ?report.error, + "task enqueue rate limit rejected control job" + ); + return Err(PersistentControlJobQueueError::TaskEnqueueRateLimited); + } + if let Some(error) = report.error { + tracing::warn!( + queue_name, + chat_id, + user_id, + control_kind = ?control_kind, + error, + "task enqueue rate limit store check failed before control job assignment" + ); + } + } + if let Some(schedule) = schedule { + self.queue.schedule_or_replace(queue_name, job, schedule); + } else { + self.queue.assign(queue_name, job); + } + // Best-effort synchronous durability for the payment-flow job: retry the + // flush a few times to ride out a transient Postgres blip so the job is + // durable before the update is acked. If every attempt fails, the 1s + // background sync worker remains the durability backstop (a crash inside + // that residual window would re-deliver the update, which is idempotent). + if let Some(journal) = &self.db_journal { + let mut last_error = None; + for attempt in 0..PAYMENT_CONTROL_FLUSH_ATTEMPTS { + match journal.flush_dirty().await { + Ok(_) => { + last_error = None; + break; + } + Err(error) => { + last_error = Some(error); + if attempt + 1 < PAYMENT_CONTROL_FLUSH_ATTEMPTS { + tokio::time::sleep(PAYMENT_CONTROL_FLUSH_RETRY_DELAY).await; + } + } + } + } + if let Some(error) = last_error { + tracing::error!( + queue_name, + control_kind = ?control_kind, + %error, + "control job not durably flushed after retries; relying on background sync worker" + ); + } + } + if let Some(rate_limit) = rate_limit { + let report = rate_limit + .grow_task_enqueue_rate_limit(chat_id, user_id, OffsetDateTime::now_utc()) + .await; + for error in report.save_errors { + tracing::warn!( + queue_name, + chat_id, + user_id, + control_kind = ?control_kind, + error, + "task enqueue rate limit store save failed after control job assignment" + ); + } + } + Ok(()) + } +} + +impl ControlJobQueue for PersistentControlJobQueue { + type Error = PersistentControlJobQueueError; + + fn assign_control_job<'a>( + &'a self, + queue_name: &'static str, + job: StatelessJobItem, + ) -> ControlJobEnqueueFuture<'a, Self::Error> { + Box::pin(self.assign_control_job_with_schedule(queue_name, job, None)) + } + + fn schedule_control_job<'a>( + &'a self, + queue_name: &'static str, + job: StatelessJobItem, + schedule: TaskQueueSchedule, + ) -> ControlJobEnqueueFuture<'a, Self::Error> { + Box::pin(self.assign_control_job_with_schedule(queue_name, job, Some(schedule))) + } +} + +impl SharedControlJobWorkerQueue for PersistentControlJobQueue { + type Error = PersistentControlJobQueueError; + + fn dequeue_shared_control_job_matching<'a>( + &'a self, + queue_name: &'static str, + worker_id: &'static str, + predicate: fn(&StatelessJobItem) -> bool, + ) -> ControlJobQueueFuture<'a, Option, Self::Error> { + Box::pin(async move { + Ok(self + .queue + .dequeue_matching(queue_name, worker_id, OffsetDateTime::now_utc(), predicate) + .map(|item| ControlJobWorkItem { + id: item.id, + job: item.job, + })) + }) + } + + fn complete_shared_control_job<'a>( + &'a self, + job_id: i64, + ) -> ControlJobQueueFuture<'a, (), Self::Error> { + Box::pin(async move { + self.queue + .complete(job_id, OffsetDateTime::now_utc()) + .map_err(control_queue_error)?; + Ok(()) + }) + } + + fn wait_for_shared_control_delivery<'a>( + &'a self, + job_id: i64, + ) -> ControlJobQueueFuture<'a, (), Self::Error> { + Box::pin(async move { + self.queue + .wait_for_delivery(job_id) + .map_err(control_queue_error)?; + if let Some(journal) = &self.db_journal { + journal.flush_job(job_id).await?; + } + Ok(()) + }) + } + + fn fail_shared_control_job<'a>( + &'a self, + job_id: i64, + error: &'a str, + ) -> ControlJobQueueFuture<'a, (), Self::Error> { + Box::pin(async move { + self.queue + .fail(job_id, error, OffsetDateTime::now_utc()) + .map_err(control_queue_error)?; + Ok(()) + }) + } +} + +#[cfg(test)] +impl Default for InMemoryControlJobQueue { + fn default() -> Self { + Self::new() + } +} + +#[cfg(test)] +impl ControlJobQueue for InMemoryControlJobQueue { + type Error = InMemoryControlJobQueueError; + + fn assign_control_job<'a>( + &'a self, + queue_name: &'static str, + job: StatelessJobItem, + ) -> ControlJobEnqueueFuture<'a, Self::Error> { + Box::pin(async move { + self.queue.assign(queue_name, job); + Ok(()) + }) + } + + fn schedule_control_job<'a>( + &'a self, + queue_name: &'static str, + job: StatelessJobItem, + schedule: TaskQueueSchedule, + ) -> ControlJobEnqueueFuture<'a, Self::Error> { + Box::pin(async move { + self.queue.schedule_or_replace(queue_name, job, schedule); + Ok(()) + }) + } +} + +#[cfg(test)] +impl SharedControlJobWorkerQueue for InMemoryControlJobQueue { + type Error = InMemoryControlJobQueueError; + + fn dequeue_shared_control_job_matching<'a>( + &'a self, + queue_name: &'static str, + worker_id: &'static str, + predicate: fn(&StatelessJobItem) -> bool, + ) -> ControlJobQueueFuture<'a, Option, Self::Error> { + Box::pin(async move { + Ok(self + .queue + .dequeue_matching(queue_name, worker_id, OffsetDateTime::now_utc(), predicate) + .map(|item| ControlJobWorkItem { + id: item.id, + job: item.job, + })) + }) + } + + fn complete_shared_control_job<'a>( + &'a self, + job_id: i64, + ) -> ControlJobQueueFuture<'a, (), Self::Error> { + Box::pin(async move { + self.queue + .complete(job_id, OffsetDateTime::now_utc()) + .map_err(control_queue_error)?; + Ok(()) + }) + } + + fn wait_for_shared_control_delivery<'a>( + &'a self, + job_id: i64, + ) -> ControlJobQueueFuture<'a, (), Self::Error> { + Box::pin(async move { + self.queue + .wait_for_delivery(job_id) + .map_err(control_queue_error)?; + Ok(()) + }) + } + + fn fail_shared_control_job<'a>( + &'a self, + job_id: i64, + error: &'a str, + ) -> ControlJobQueueFuture<'a, (), Self::Error> { + Box::pin(async move { + self.queue + .fail(job_id, error, OffsetDateTime::now_utc()) + .map_err(control_queue_error)?; + Ok(()) + }) + } +} + +fn control_queue_error(error: openplotva_taskman::TaskQueueError) -> InMemoryControlJobQueueError { + match error { + openplotva_taskman::TaskQueueError::JobNotFound(job_id) => { + InMemoryControlJobQueueError::JobNotFound(job_id) + } + error => InMemoryControlJobQueueError::Taskman(error.to_string()), + } +} + +#[cfg(test)] +fn control_records_from_taskman(records: &[TaskQueueRecord]) -> Vec { + records + .iter() + .map(|record| InMemoryControlJobRecord { + id: record.id, + queue_name: record.queue_name.clone(), + status: control_status_from_taskman(record.status), + job: record.job.clone(), + error: record.error.clone(), + }) + .collect() +} + +#[cfg(test)] +fn control_status_from_taskman(status: openplotva_taskman::JobStatus) -> InMemoryControlJobStatus { + use openplotva_taskman::JobStatus; + match status { + JobStatus::Pending => InMemoryControlJobStatus::Pending, + JobStatus::Processing | JobStatus::WaitingDelivery => InMemoryControlJobStatus::Processing, + JobStatus::Completed => InMemoryControlJobStatus::Completed, + JobStatus::Failed | JobStatus::Cancelled => InMemoryControlJobStatus::Failed, + } +} + use std::{fmt, future::Future, pin::Pin, time::Duration}; use openplotva_taskman::{ @@ -8,11 +550,11 @@ use openplotva_telegram::DispatcherQueue; use time::OffsetDateTime; use crate::{ - checkin::{CheckinControlJobEffects, CheckinControlJobOutcome}, + checkin::CheckinControlJobEffects, members::{MemberStateControlJobEffects, MemberStateControlJobExecution}, payments::{ - PaymentControlJobReport, SharedControlJobWorkerQueue, SuccessfulPaymentStore, - execute_payment_control_job_at, payment_control_job_failure_message, + PaymentControlJobReport, SuccessfulPaymentStore, execute_payment_control_job_at, + payment_control_job_failure_message, }, settings::{ GroupSettingsControlJobEffects, NewMembersFollowupControlJobEffects, @@ -43,7 +585,7 @@ pub enum ControlJobExecution { /// Chat-admin/member sync executor result. MemberState(MemberStateControlJobExecution), /// Check-in game executor result. - Checkin(CheckinControlJobOutcome), + Checkin, } /// Result returned by one control-kind executor. @@ -51,6 +593,7 @@ pub enum ControlJobExecution { pub enum ControlJobExecutionResult { /// Executor completed and the queue row should be marked completed. Completed(ControlJobExecution), + WaitingDelivery(crate::telegram_outbox::QueuedBatchReceipt), /// Executor failed and the queue row should be marked failed. Failed { /// Optional partial execution details. @@ -73,6 +616,7 @@ pub struct ControlJobWorkerReport { pub execution: Option, /// The job was finalized as completed. pub completed: bool, + pub waiting_delivery: bool, /// The job was finalized as failed. pub failed: bool, /// Queue dequeue failed. @@ -93,6 +637,7 @@ pub struct ControlJobWorkerRunReport { pub dequeued: u64, /// Number of jobs completed. pub completed: u64, + pub waiting_delivery: u64, /// Number of jobs failed. pub failed: u64, /// Number of queue dequeue errors. @@ -105,6 +650,7 @@ pub trait ControlJobExecutorRegistry { /// Execute one already-decoded control job. fn execute_control_job<'a>( &'a self, + job_id: i64, params: &'a ControlJobParams, now: OffsetDateTime, ) -> ControlJobExecutorFuture<'a>; @@ -183,6 +729,7 @@ where { fn execute_control_job<'a>( &'a self, + job_id: i64, params: &'a ControlJobParams, now: OffsetDateTime, ) -> ControlJobExecutorFuture<'a> { @@ -285,15 +832,18 @@ where MemberStateControlJobExecution::SupersededByProjectionMaterializer, )) } - ControlKind::Checkin => match self.checkin_effects.run_checkin_game(params).await { - Ok(()) => ControlJobExecutionResult::Completed(ControlJobExecution::Checkin( - CheckinControlJobOutcome::Completed, - )), - Err(error) => ControlJobExecutionResult::Failed { - execution: None, - error: error.to_string(), - }, - }, + ControlKind::Checkin => { + match self.checkin_effects.run_checkin_game(job_id, params).await { + Ok(Some(receipt)) => ControlJobExecutionResult::WaitingDelivery(receipt), + Ok(None) => { + ControlJobExecutionResult::Completed(ControlJobExecution::Checkin) + } + Err(error) => ControlJobExecutionResult::Failed { + execution: None, + error: error.to_string(), + }, + } + } kind => unsupported_control_job_kind(kind), } }) @@ -371,7 +921,7 @@ where let result = match tokio::time::timeout( control_job_timeout(params.data.kind), - registry.execute_control_job(¶ms, now), + registry.execute_control_job(item.id, ¶ms, now), ) .await { @@ -394,6 +944,12 @@ where Err(error) => report.status_error = Some(error.to_string()), } } + ControlJobExecutionResult::WaitingDelivery(_) => { + match queue.wait_for_shared_control_delivery(item.id).await { + Ok(()) => report.waiting_delivery = true, + Err(error) => report.status_error = Some(error.to_string()), + } + } ControlJobExecutionResult::Failed { execution, error } => { report.execution = execution; report.execution_error = Some(error.clone()); @@ -464,6 +1020,9 @@ impl ControlJobWorkerRunReport { if tick.completed { self.completed += 1; } + if tick.waiting_delivery { + self.waiting_delivery += 1; + } if tick.failed { self.failed += 1; } @@ -516,14 +1075,15 @@ fn is_control_job(job: &StatelessJobItem) -> bool { impl ControlJobExecutorRegistry for F where - F: for<'a> Fn(&'a ControlJobParams, OffsetDateTime) -> ControlJobExecutorFuture<'a> + Sync, + F: for<'a> Fn(i64, &'a ControlJobParams, OffsetDateTime) -> ControlJobExecutorFuture<'a> + Sync, { fn execute_control_job<'a>( &'a self, + job_id: i64, params: &'a ControlJobParams, now: OffsetDateTime, ) -> ControlJobExecutorFuture<'a> { - self(params, now) + self(job_id, params, now) } } @@ -531,6 +1091,7 @@ impl fmt::Display for ControlJobExecutionResult { fn fmt(&self, formatter: &mut fmt::Formatter<'_>) -> fmt::Result { match self { Self::Completed(_) => formatter.write_str("completed"), + Self::WaitingDelivery(_) => formatter.write_str("waiting_delivery"), Self::Failed { error, .. } => error.fmt(formatter), } } @@ -549,22 +1110,23 @@ mod tests { }; use super::*; - use crate::payments::{InMemoryPaymentControlJobStatus, PaymentControlJobQueue}; + + use crate::control_jobs::{ControlJobQueue, InMemoryControlJobStatus}; #[tokio::test] async fn unified_control_worker_dequeues_next_pending_control_job_by_go_order() -> Result<(), Box> { - let queue = crate::payments::InMemoryPaymentControlJobQueue::new(); + let queue = crate::control_jobs::InMemoryControlJobQueue::new(); let now = OffsetDateTime::from_unix_timestamp(1_779_193_800)?; queue - .assign_payment_control_job( + .assign_control_job( CONTROL_QUEUE_NAME, control_job(ControlKind::Translate, now, 0), ) .await?; let low_id = queue.snapshot().last().expect("low job").id; queue - .assign_payment_control_job( + .assign_control_job( CONTROL_QUEUE_NAME, control_job( ControlKind::Checkin, @@ -584,22 +1146,17 @@ mod tests { assert_eq!(report.job_id, Some(high_id)); assert_eq!(report.kind, Some(ControlKind::Checkin)); - assert_eq!( - report.execution, - Some(ControlJobExecution::Checkin( - CheckinControlJobOutcome::Completed - )) - ); + assert_eq!(report.execution, Some(ControlJobExecution::Checkin)); assert!(report.completed); assert_eq!(*seen.lock().expect("seen"), vec![ControlKind::Checkin]); let snapshot = queue.snapshot(); assert_eq!( snapshot_record(&snapshot, high_id).status, - InMemoryPaymentControlJobStatus::Completed + InMemoryControlJobStatus::Completed ); assert_eq!( snapshot_record(&snapshot, low_id).status, - InMemoryPaymentControlJobStatus::Pending + InMemoryControlJobStatus::Pending ); Ok(()) } @@ -607,16 +1164,16 @@ mod tests { #[tokio::test] async fn transactional_control_job_overtakes_older_membership_sync() -> Result<(), Box> { - let queue = crate::payments::InMemoryPaymentControlJobQueue::new(); + let queue = crate::control_jobs::InMemoryControlJobQueue::new(); let now = OffsetDateTime::from_unix_timestamp(1_779_193_800)?; queue - .assign_payment_control_job( + .assign_control_job( CONTROL_QUEUE_NAME, control_job(ControlKind::ChatMemberSync, now, HIGH_PRIORITY), ) .await?; queue - .assign_payment_control_job( + .assign_control_job( CONTROL_QUEUE_NAME, control_job( ControlKind::SuccessfulPayment, @@ -641,12 +1198,52 @@ mod tests { Ok(()) } + #[tokio::test] + async fn queued_control_result_waits_for_delivery_without_completing() + -> Result<(), Box> { + fn queued( + job_id: i64, + _params: &ControlJobParams, + _now: OffsetDateTime, + ) -> ControlJobExecutorFuture<'_> { + Box::pin(async move { + ControlJobExecutionResult::WaitingDelivery( + crate::telegram_outbox::QueuedBatchReceipt::durable_outbox( + crate::checkin::checkin_delivery_source(job_id), + vec!["checkin-part".into()], + false, + ), + ) + }) + } + let queue = crate::control_jobs::InMemoryControlJobQueue::new(); + let now = OffsetDateTime::now_utc(); + queue + .assign_control_job( + CONTROL_QUEUE_NAME, + control_job(ControlKind::Checkin, now, HIGH_PRIORITY), + ) + .await?; + let report = process_control_job_once_at(&queue, &queued, now).await; + assert!(report.waiting_delivery); + assert!(!report.completed); + assert!(!report.failed); + let records = queue.taskman_records(); + assert_eq!( + records[0].status, + openplotva_taskman::JobStatus::WaitingDelivery + ); + assert_eq!(records[0].result_message_id, None); + assert_eq!(records[0].completed_at, None); + Ok(()) + } + #[tokio::test] async fn unified_control_worker_fails_unknown_control_kind() -> Result<(), Box> { - let queue = crate::payments::InMemoryPaymentControlJobQueue::new(); + let queue = crate::control_jobs::InMemoryControlJobQueue::new(); let now = OffsetDateTime::from_unix_timestamp(1_779_193_800)?; queue - .assign_payment_control_job( + .assign_control_job( CONTROL_QUEUE_NAME, control_job(ControlKind::Unknown, now, HIGH_PRIORITY), ) @@ -665,7 +1262,7 @@ mod tests { ); let snapshot = queue.snapshot(); let record = snapshot_record(&snapshot, job_id); - assert_eq!(record.status, InMemoryPaymentControlJobStatus::Failed); + assert_eq!(record.status, InMemoryControlJobStatus::Failed); assert_eq!( record.error.as_deref(), Some("unsupported control job kind: Unknown") @@ -725,6 +1322,7 @@ mod tests { impl ControlJobExecutorRegistry for RecordingRegistry { fn execute_control_job<'a>( &'a self, + _job_id: i64, params: &'a ControlJobParams, _now: OffsetDateTime, ) -> ControlJobExecutorFuture<'a> { @@ -743,9 +1341,9 @@ mod tests { ControlKind::Translate => ControlJobExecutionResult::Completed( ControlJobExecution::Translate(TranslateControlJobOutcome::Sent), ), - ControlKind::Checkin => ControlJobExecutionResult::Completed( - ControlJobExecution::Checkin(CheckinControlJobOutcome::Completed), - ), + ControlKind::Checkin => { + ControlJobExecutionResult::Completed(ControlJobExecution::Checkin) + } kind => unsupported_control_job_kind(kind), } }) @@ -755,11 +1353,11 @@ mod tests { #[tokio::test(start_paused = true)] async fn worker_drains_ready_control_jobs_without_poll_interval_between_them() -> Result<(), Box> { - let queue = crate::payments::InMemoryPaymentControlJobQueue::new(); + let queue = crate::control_jobs::InMemoryControlJobQueue::new(); let now = OffsetDateTime::from_unix_timestamp(1_779_193_800)?; for kind in [ControlKind::Translate, ControlKind::Checkin] { queue - .assign_payment_control_job(CONTROL_QUEUE_NAME, control_job(kind, now, 0)) + .assign_control_job(CONTROL_QUEUE_NAME, control_job(kind, now, 0)) .await?; } let notify = Arc::new(tokio::sync::Notify::new()); @@ -805,12 +1403,15 @@ mod tests { } fn snapshot_record( - snapshot: &[crate::payments::InMemoryPaymentControlJobRecord], + snapshot: &[crate::control_jobs::InMemoryControlJobRecord], id: i64, - ) -> &crate::payments::InMemoryPaymentControlJobRecord { + ) -> &crate::control_jobs::InMemoryControlJobRecord { snapshot .iter() .find(|record| record.id == id) .expect("snapshot record") } } + +#[cfg(test)] +pub(crate) mod test_support; diff --git a/crates/openplotva-app/src/control_jobs/test_support.rs b/crates/openplotva-app/src/control_jobs/test_support.rs new file mode 100644 index 00000000..c3be03a0 --- /dev/null +++ b/crates/openplotva-app/src/control_jobs/test_support.rs @@ -0,0 +1,190 @@ +use super::*; +use crate::checkin::*; +use crate::members::*; +use crate::payments::*; +use crate::settings::*; +use crate::translate::*; +use openplotva_core::UserState; +use openplotva_storage::*; + +pub struct UnusedControlPorts; + +impl SuccessfulPaymentStore for UnusedControlPorts { + type Error = std::convert::Infallible; + fn upsert_payment_user<'a>( + &'a self, + _user: &'a UserState, + ) -> PaymentStoreFuture<'a, (), Self::Error> { + panic!("unexpected control executor port") + } + fn create_subscription<'a>( + &'a self, + _subscription: SubscriptionCreate<'a>, + ) -> PaymentStoreFuture<'a, SubscriptionRecord, Self::Error> { + panic!("unexpected control executor port") + } + fn get_subscription_by_telegram_payment_charge_id<'a>( + &'a self, + _telegram_payment_charge_id: &'a str, + ) -> PaymentStoreFuture<'a, Option, Self::Error> { + panic!("unexpected control executor port") + } + fn get_vip_summary_by_user<'a>( + &'a self, + _user_id: i64, + ) -> PaymentStoreFuture<'a, Option, Self::Error> { + panic!("unexpected control executor port") + } + fn get_vip_event_by_subscription_id<'a>( + &'a self, + _subscription_id: i64, + ) -> PaymentStoreFuture<'a, Option, Self::Error> { + panic!("unexpected control executor port") + } + fn create_donation<'a>( + &'a self, + _donation: DonationCreate<'a>, + ) -> PaymentStoreFuture<'a, DonationRecord, Self::Error> { + panic!("unexpected control executor port") + } + fn get_donation_by_telegram_payment_charge_id<'a>( + &'a self, + _telegram_payment_charge_id: &'a str, + ) -> PaymentStoreFuture<'a, Option, Self::Error> { + panic!("unexpected control executor port") + } + fn create_vip_event<'a>( + &'a self, + _event: VipEventCreate<'a>, + ) -> PaymentStoreFuture<'a, VipEventRecord, Self::Error> { + panic!("unexpected control executor port") + } + fn is_duplicate_insert_no_row(_error: &Self::Error) -> bool { + panic!("unexpected control executor port") + } +} + +impl PaymentInvoiceEffects for UnusedControlPorts { + type Error = std::convert::Infallible; + fn create_subscription_invoice_link<'a>( + &'a self, + _request: &'a openplotva_telegram::SubscriptionInvoiceLinkRequest, + ) -> PaymentInvoiceLinkFuture<'a, Self::Error> { + panic!("unexpected control executor port") + } + fn subscription_price_override(&self, _user_id: i64) -> Option { + panic!("unexpected control executor port") + } + fn create_donation_invoice_link<'a>( + &'a self, + _request: &'a openplotva_telegram::DonationInvoiceLinkRequest, + ) -> PaymentInvoiceLinkFuture<'a, Self::Error> { + panic!("unexpected control executor port") + } + fn send_invoice_button_message<'a>( + &'a self, + _message: InvoiceButtonMessage, + ) -> PaymentEffectsFuture<'a, Self::Error> { + panic!("unexpected control executor port") + } + fn send_invoice_error_text<'a>( + &'a self, + _chat_id: i64, + _reply_to_message_id: i32, + _text: &'a str, + _parse_mode: &'a str, + ) -> PaymentEffectsFuture<'a, Self::Error> { + panic!("unexpected control executor port") + } +} + +impl SuccessfulPaymentEffects for UnusedControlPorts { + type Error = std::convert::Infallible; + fn send_text<'a>( + &'a self, + _message: PaymentTextMessage<'a>, + ) -> PaymentEffectsFuture<'a, Self::Error> { + panic!("unexpected control executor port") + } + fn invalidate_vip_cache<'a>(&'a self, _user_id: i64) -> PaymentEffectsFuture<'a, Self::Error> { + panic!("unexpected control executor port") + } + fn support_url(&self) -> &str { + panic!("unexpected control executor port") + } +} + +impl GroupSettingsControlJobEffects for UnusedControlPorts { + type Error = std::convert::Infallible; + fn can_open_group_settings<'a>( + &'a self, + _chat_id: i64, + _user_id: i64, + ) -> GroupSettingsControlJobFuture<'a, bool, Self::Error> { + panic!("unexpected control executor port") + } + fn sync_chat_admins<'a>(&'a self, _chat_id: i64) -> GroupSettingsSyncFuture<'a> { + panic!("unexpected control executor port") + } +} + +impl NewMembersFollowupControlJobEffects for UnusedControlPorts { + fn enable_chat_communication<'a>(&'a self, _chat_id: i64) -> NewMembersFollowupFuture<'a, ()> { + panic!("unexpected control executor port") + } + fn is_chat_blocked<'a>(&'a self, _chat_id: i64) -> NewMembersFollowupFuture<'a, bool> { + panic!("unexpected control executor port") + } + fn try_send_greeting_for_join_wave<'a>( + &'a self, + _greeting: NewMembersGreeting, + ) -> NewMembersFollowupFuture<'a, ()> { + panic!("unexpected control executor port") + } +} + +impl TextTranslator for UnusedControlPorts { + type Error = std::convert::Infallible; + fn translate_text<'a>( + &'a self, + _text: &'a str, + _target_lang: &'a str, + ) -> TranslateProviderFuture<'a, Self::Error> { + panic!("unexpected control executor port") + } +} + +impl TranslateEffects for UnusedControlPorts { + type Error = std::convert::Infallible; + fn send_translate_notice<'a>( + &'a self, + _message: TranslateCommandMessage, + _text: String, + _delete_after: Duration, + ) -> TranslateEffectFuture<'a, Self::Error> { + panic!("unexpected control executor port") + } + fn send_translation_result<'a>( + &'a self, + _plan: TranslateResultPlan, + ) -> TranslateEffectFuture<'a, Self::Error> { + panic!("unexpected control executor port") + } +} + +impl MemberStateControlJobEffects for UnusedControlPorts { + fn sync_chat_admins<'a>(&'a self, _chat_id: i64) -> MemberStateControlJobFuture<'a> { + panic!("unexpected control executor port") + } +} + +impl CheckinControlJobEffects for UnusedControlPorts { + type Error = std::convert::Infallible; + fn run_checkin_game<'a>( + &'a self, + _job_id: i64, + _params: &'a ControlJobParams, + ) -> CheckinControlJobFuture<'a, Self::Error> { + panic!("unexpected control executor port") + } +} diff --git a/crates/openplotva-app/src/delete_drawing.rs b/crates/openplotva-app/src/delete_drawing.rs index a30a6050..c2322a22 100644 --- a/crates/openplotva-app/src/delete_drawing.rs +++ b/crates/openplotva-app/src/delete_drawing.rs @@ -1,28 +1,36 @@ //! App-level fetcher `/delete_drawing` command behavior. +#[cfg(test)] +use crate::callbacks::DeleteMemberFuture; +use crate::callbacks::{ + DeleteCallbackEffects, DeleteMemberApi, can_delete_owned_content, try_execute_delete_method, +}; +#[cfg(test)] +use carapax::types::ChatMember as TelegramChatMember; +#[cfg(test)] +use openplotva_telegram::TelegramOutboundMethod; use std::{fmt, future::Future, pin::Pin, sync::Arc, time::Duration}; use carapax::types::{ - CallbackQuery as TelegramCallbackQuery, ChatMember as TelegramChatMember, - MaybeInaccessibleMessage, Message as TelegramMessage, Update as TelegramUpdate, - UpdateType as TelegramUpdateType, + CallbackQuery as TelegramCallbackQuery, MaybeInaccessibleMessage, Message as TelegramMessage, + Update as TelegramUpdate, UpdateType as TelegramUpdateType, }; use openplotva_telegram::{ CallbackActionParse, CallbackAnswerRequest, ChatRef, DeleteMessageRequest, DispatcherQueue, EditCaptionMessageRequest, EditReplyMarkupMessageRequest, EditTextMessageRequest, - OutboundBuildError, ReplyMarkup, ReplyMessageRef, TELEGRAM_PARSE_MODE_HTML, - TelegramOutboundMethod, TextMessageRequest, build_callback_answer_method, - build_delete_drawing_confirm_keyboard, build_delete_drawing_frame_confirm_keyboard, - build_delete_drawing_frame_picker_keyboard, build_delete_drawing_initial_keyboard, - build_delete_message_method, build_edit_caption_message_method, - build_edit_reply_markup_message_method, build_edit_text_message_method, - build_get_chat_member_method, parse_callback_action, parse_callback_i64, + OutboundBuildError, ReplyMarkup, ReplyMessageRef, TELEGRAM_PARSE_MODE_HTML, TextMessageRequest, + build_callback_answer_method, build_delete_drawing_confirm_keyboard, + build_delete_drawing_frame_confirm_keyboard, build_delete_drawing_frame_picker_keyboard, + build_delete_drawing_initial_keyboard, build_delete_message_method, + build_edit_caption_message_method, build_edit_reply_markup_message_method, + build_edit_text_message_method, parse_callback_action, parse_callback_i64, }; use openplotva_updates::parse_leading_bot_command; use thiserror::Error; use time::OffsetDateTime; use crate::updates::UpdateHandler; + use crate::virtual_messages::{ QueueTextRequest, VirtualIdFactory, monotonic_virtual_id_factory, queue_text_message_parts, }; @@ -162,37 +170,6 @@ impl DeleteDrawingCallbackStore for openplotva_storage::RedisLastGenerationStore } } -/// Boxed future returned by Telegram membership APIs. -pub type DeleteDrawingMemberFuture<'a, E> = - Pin> + Send + 'a>>; - -pub trait DeleteDrawingMemberApi { - /// Concrete API error. - type Error: fmt::Display + Send + Sync + 'static; - - /// Fetch one Telegram chat member. - fn get_chat_member<'a>( - &'a self, - chat_id: i64, - user_id: i64, - ) -> DeleteDrawingMemberFuture<'a, Self::Error>; -} - -impl DeleteDrawingMemberApi for openplotva_telegram::TelegramClient { - type Error = carapax::api::ExecuteError; - - fn get_chat_member<'a>( - &'a self, - chat_id: i64, - user_id: i64, - ) -> DeleteDrawingMemberFuture<'a, Self::Error> { - Box::pin(async move { - self.execute(build_get_chat_member_method(chat_id, user_id)) - .await - }) - } -} - /// Boxed future returned by VIP status lookups. pub type DeleteDrawingVipFuture<'a> = Pin + Send + 'a>>; @@ -275,28 +252,6 @@ impl DeleteDrawingCommandEffects for DeleteDrawingCommandDispatcherEffects { } } -pub trait DeleteDrawingCallbackEffects { - /// Concrete effect error. - type Error: fmt::Display + Send + Sync + 'static; - - /// Execute one direct Telegram Bot API method. - fn execute_delete_drawing_method<'a>( - &'a self, - method: TelegramOutboundMethod, - ) -> DeleteDrawingEffectFuture<'a, Self::Error>; -} - -impl DeleteDrawingCallbackEffects for openplotva_telegram::TelegramClient { - type Error = carapax::api::ExecuteError; - - fn execute_delete_drawing_method<'a>( - &'a self, - method: TelegramOutboundMethod, - ) -> DeleteDrawingEffectFuture<'a, Self::Error> { - Box::pin(async move { method.execute_with(self).await.map(|_| ()) }) - } -} - /// Result of one decoded update through the delete-drawing wrapper. #[derive(Clone, Debug, Eq, PartialEq)] pub enum DeleteDrawingCommandOutcome { @@ -432,9 +387,9 @@ impl UpdateHandler for DeleteDrawingCallbackUpdateHandler where Store: DeleteDrawingCallbackStore + Send + Sync, - MemberApi: DeleteDrawingMemberApi + Send + Sync, + MemberApi: DeleteMemberApi + Send + Sync, Vip: DeleteDrawingVipStatus + Send + Sync, - Effects: DeleteDrawingCallbackEffects + Send + Sync, + Effects: DeleteCallbackEffects + Send + Sync, Next: UpdateHandler + Send + Sync, { type Error = DeleteDrawingCallbackError; @@ -509,9 +464,9 @@ pub async fn handle_delete_drawing_callback_update_or_else< ) -> Result where Store: DeleteDrawingCallbackStore + Sync, - MemberApi: DeleteDrawingMemberApi + Sync, + MemberApi: DeleteMemberApi + Sync, Vip: DeleteDrawingVipStatus + Sync, - Effects: DeleteDrawingCallbackEffects + Sync, + Effects: DeleteCallbackEffects + Sync, HandleFn: FnOnce(TelegramUpdate) -> HandleFuture, HandleFuture: Future>, HandleError: fmt::Display, @@ -558,11 +513,11 @@ pub async fn handle_delete_drawing_callback( ) -> DeleteDrawingCallbackOutcome where Store: DeleteDrawingCallbackStore + Sync, - MemberApi: DeleteDrawingMemberApi + Sync, + MemberApi: DeleteMemberApi + Sync, Vip: DeleteDrawingVipStatus + Sync, - Effects: DeleteDrawingCallbackEffects + Sync, + Effects: DeleteCallbackEffects + Sync, { - if !is_authorized_for_delete_drawing(member_api, query, data.user_id, data.chat_id).await { + if !can_delete_owned_content(member_api, query, data.user_id, data.chat_id).await { try_answer_callback( effects, &query.id, @@ -706,7 +661,7 @@ async fn handle_delete_drawing_init( where Store: DeleteDrawingCallbackStore + Sync, Vip: DeleteDrawingVipStatus + Sync, - Effects: DeleteDrawingCallbackEffects + Sync, + Effects: DeleteCallbackEffects + Sync, { let Some(generation) = load_callback_generation(store, data, true).await else { try_answer_callback( @@ -755,7 +710,7 @@ async fn handle_delete_drawing_confirm( ) -> DeleteDrawingCallbackOutcome where Store: DeleteDrawingCallbackStore + Sync, - Effects: DeleteDrawingCallbackEffects + Sync, + Effects: DeleteCallbackEffects + Sync, { let Some(generation) = load_callback_generation(store, data, false).await else { try_answer_callback( @@ -800,7 +755,7 @@ async fn handle_delete_drawing_frame_pick( ) -> DeleteDrawingCallbackOutcome where Vip: DeleteDrawingVipStatus + Sync, - Effects: DeleteDrawingCallbackEffects + Sync, + Effects: DeleteCallbackEffects + Sync, { if !vip.is_delete_drawing_vip(i64::from(query.from.id)).await { try_answer_callback( @@ -843,7 +798,7 @@ async fn handle_delete_drawing_frame_confirm( where Store: DeleteDrawingCallbackStore + Sync, Vip: DeleteDrawingVipStatus + Sync, - Effects: DeleteDrawingCallbackEffects + Sync, + Effects: DeleteCallbackEffects + Sync, { if !vip.is_delete_drawing_vip(i64::from(query.from.id)).await { try_answer_callback( @@ -970,7 +925,7 @@ async fn handle_delete_drawing_close( control_message: DeleteDrawingControlMessage, ) -> DeleteDrawingCallbackOutcome where - Effects: DeleteDrawingCallbackEffects + Sync, + Effects: DeleteCallbackEffects + Sync, { try_delete_message( effects, @@ -1045,7 +1000,7 @@ async fn try_answer_callback( show_alert: bool, context: &'static str, ) where - Effects: DeleteDrawingCallbackEffects + Sync, + Effects: DeleteCallbackEffects + Sync, { let request = CallbackAnswerRequest { callback_query_id: query_id.to_owned(), @@ -1054,7 +1009,7 @@ async fn try_answer_callback( url: String::new(), cache_time: 0, }; - try_execute_delete_drawing_method( + try_execute_delete_method( effects, build_callback_answer_method(&request).into(), context, @@ -1067,10 +1022,10 @@ async fn try_edit_reply_markup( request: EditReplyMarkupMessageRequest, context: &'static str, ) where - Effects: DeleteDrawingCallbackEffects + Sync, + Effects: DeleteCallbackEffects + Sync, { match build_edit_reply_markup_message_method(&request) { - Ok(method) => try_execute_delete_drawing_method(effects, method.into(), context).await, + Ok(method) => try_execute_delete_method(effects, method.into(), context).await, Err(error) => tracing::warn!(message = %error, "failed to build delete drawing method"), } } @@ -1080,10 +1035,10 @@ async fn try_edit_text( request: EditTextMessageRequest, context: &'static str, ) where - Effects: DeleteDrawingCallbackEffects + Sync, + Effects: DeleteCallbackEffects + Sync, { match build_edit_text_message_method(&request) { - Ok(method) => try_execute_delete_drawing_method(effects, method.into(), context).await, + Ok(method) => try_execute_delete_method(effects, method.into(), context).await, Err(error) => tracing::warn!(message = %error, "failed to build delete drawing method"), } } @@ -1093,10 +1048,10 @@ async fn try_edit_caption( request: EditCaptionMessageRequest, context: &'static str, ) where - Effects: DeleteDrawingCallbackEffects + Sync, + Effects: DeleteCallbackEffects + Sync, { match build_edit_caption_message_method(&request) { - Ok(method) => try_execute_delete_drawing_method(effects, method.into(), context).await, + Ok(method) => try_execute_delete_method(effects, method.into(), context).await, Err(error) => tracing::warn!(message = %error, "failed to build delete drawing method"), } } @@ -1107,75 +1062,17 @@ async fn try_delete_message( message_id: i64, context: &'static str, ) where - Effects: DeleteDrawingCallbackEffects + Sync, + Effects: DeleteCallbackEffects + Sync, { match build_delete_message_method(&DeleteMessageRequest { chat_id, message_id, }) { - Ok(method) => try_execute_delete_drawing_method(effects, method.into(), context).await, + Ok(method) => try_execute_delete_method(effects, method.into(), context).await, Err(error) => tracing::warn!(message = %error, "failed to build delete drawing method"), } } -async fn try_execute_delete_drawing_method( - effects: &Effects, - method: TelegramOutboundMethod, - context: &'static str, -) where - Effects: DeleteDrawingCallbackEffects + Sync, -{ - let method_name = method.method_name(); - if let Err(error) = effects.execute_delete_drawing_method(method).await { - tracing::warn!( - message = %error, - method = method_name, - context, - "delete drawing callback Telegram side effect failed" - ); - } -} - -async fn is_authorized_for_delete_drawing( - member_api: &MemberApi, - query: &TelegramCallbackQuery, - data_user_id: i64, - data_chat_id: i64, -) -> bool -where - MemberApi: DeleteDrawingMemberApi + Sync, -{ - let requester_id = i64::from(query.from.id); - if requester_id == data_user_id { - return true; - } - - match member_api.get_chat_member(data_chat_id, requester_id).await { - Ok(member) => telegram_member_can_delete_drawing(&member), - Err(error) => { - tracing::warn!( - message = %error, - chat_id = data_chat_id, - user_id = requester_id, - "failed to authorize delete drawing callback" - ); - false - } - } -} - -#[must_use] -pub fn telegram_member_can_delete_drawing(member: &TelegramChatMember) -> bool { - match member { - TelegramChatMember::Creator(_) => true, - TelegramChatMember::Administrator(admin) => admin.can_delete_messages, - TelegramChatMember::Kicked(_) - | TelegramChatMember::Left(_) - | TelegramChatMember::Member { .. } - | TelegramChatMember::Restricted(_) => false, - } -} - #[must_use] pub fn plan_delete_drawing_frame( generation: &LastGeneration, @@ -2277,10 +2174,10 @@ mod tests { } } - impl DeleteDrawingCallbackEffects for EffectsStub { + impl DeleteCallbackEffects for EffectsStub { type Error = io::Error; - fn execute_delete_drawing_method<'a>( + fn execute_delete_method<'a>( &'a self, method: TelegramOutboundMethod, ) -> DeleteDrawingEffectFuture<'a, Self::Error> { @@ -2361,14 +2258,14 @@ mod tests { } } - impl DeleteDrawingMemberApi for MemberApiStub { + impl DeleteMemberApi for MemberApiStub { type Error = io::Error; fn get_chat_member<'a>( &'a self, chat_id: i64, user_id: i64, - ) -> DeleteDrawingMemberFuture<'a, Self::Error> { + ) -> DeleteMemberFuture<'a, Self::Error> { Box::pin(async move { self.calls .lock() diff --git a/crates/openplotva-app/src/delete_lyrics.rs b/crates/openplotva-app/src/delete_lyrics.rs index 259df1c8..7f3207bb 100644 --- a/crates/openplotva-app/src/delete_lyrics.rs +++ b/crates/openplotva-app/src/delete_lyrics.rs @@ -1,17 +1,27 @@ //! App-level generated-lyrics delete callback behavior. -use std::{fmt, future::Future, pin::Pin, sync::Arc}; +#[cfg(test)] +use crate::callbacks::DeleteEffectFuture; +#[cfg(test)] +use crate::callbacks::DeleteMemberFuture; +use crate::callbacks::{ + DeleteCallbackEffects, DeleteMemberApi, can_delete_owned_content, try_execute_delete_method, +}; +#[cfg(test)] +use carapax::types::ChatMember as TelegramChatMember; +#[cfg(test)] +use openplotva_telegram::TelegramOutboundMethod; +use std::{fmt, future::Future, sync::Arc}; use carapax::types::{ - CallbackQuery as TelegramCallbackQuery, ChatMember as TelegramChatMember, - MaybeInaccessibleMessage, Update as TelegramUpdate, UpdateType as TelegramUpdateType, + CallbackQuery as TelegramCallbackQuery, MaybeInaccessibleMessage, Update as TelegramUpdate, + UpdateType as TelegramUpdateType, }; use openplotva_telegram::{ CallbackActionParse, CallbackAnswerRequest, ChatRef, DeleteMessageRequest, - EditReplyMarkupMessageRequest, TelegramOutboundMethod, build_callback_answer_method, - build_delete_message_method, build_edit_reply_markup_message_method, - build_get_chat_member_method, build_lyrics_delete_confirm_keyboard, parse_callback_action, - parse_callback_i64, + EditReplyMarkupMessageRequest, build_callback_answer_method, build_delete_message_method, + build_edit_reply_markup_message_method, build_lyrics_delete_confirm_keyboard, + parse_callback_action, parse_callback_i64, }; use thiserror::Error; @@ -20,62 +30,6 @@ use crate::updates::UpdateHandler; const DELETE_LYRICS_AUTH_ALERT_TEXT: &str = "Π£Π΄Π°Π»Π΅Π½ΠΈΠ΅ доступно Ρ‚ΠΎΠ»ΡŒΠΊΠΎ Π°Π²Ρ‚ΠΎΡ€Ρƒ ΠΈΠ»ΠΈ администраторам."; const DELETE_LYRICS_DELETED_TEXT: &str = "βœ… ВСкст ΡƒΠ΄Π°Π»Ρ‘Π½"; -/// Boxed future returned by Telegram membership APIs. -pub type DeleteLyricsMemberFuture<'a, E> = - Pin> + Send + 'a>>; - -pub trait DeleteLyricsMemberApi { - /// Concrete API error. - type Error: fmt::Display + Send + Sync + 'static; - - /// Fetch one Telegram chat member. - fn get_chat_member<'a>( - &'a self, - chat_id: i64, - user_id: i64, - ) -> DeleteLyricsMemberFuture<'a, Self::Error>; -} - -impl DeleteLyricsMemberApi for openplotva_telegram::TelegramClient { - type Error = carapax::api::ExecuteError; - - fn get_chat_member<'a>( - &'a self, - chat_id: i64, - user_id: i64, - ) -> DeleteLyricsMemberFuture<'a, Self::Error> { - Box::pin(async move { - self.execute(build_get_chat_member_method(chat_id, user_id)) - .await - }) - } -} - -/// Boxed future returned by delete-lyrics effects. -pub type DeleteLyricsEffectFuture<'a, E> = Pin> + Send + 'a>>; - -pub trait DeleteLyricsEffects { - /// Concrete effect error. - type Error: fmt::Display + Send + Sync + 'static; - - /// Execute one direct Telegram Bot API method. - fn execute_delete_lyrics_method<'a>( - &'a self, - method: TelegramOutboundMethod, - ) -> DeleteLyricsEffectFuture<'a, Self::Error>; -} - -impl DeleteLyricsEffects for openplotva_telegram::TelegramClient { - type Error = carapax::api::ExecuteError; - - fn execute_delete_lyrics_method<'a>( - &'a self, - method: TelegramOutboundMethod, - ) -> DeleteLyricsEffectFuture<'a, Self::Error> { - Box::pin(async move { method.execute_with(self).await.map(|_| ()) }) - } -} - #[derive(Clone, Debug, Eq, PartialEq)] pub struct DeleteLyricsCallbackData { pub action: String, @@ -138,8 +92,8 @@ impl DeleteLyricsCallbackUpdateHandler UpdateHandler for DeleteLyricsCallbackUpdateHandler where - MemberApi: DeleteLyricsMemberApi + Send + Sync, - Effects: DeleteLyricsEffects + Send + Sync, + MemberApi: DeleteMemberApi + Send + Sync, + Effects: DeleteCallbackEffects + Send + Sync, Next: UpdateHandler + Send + Sync, { type Error = DeleteLyricsCallbackError; @@ -169,8 +123,8 @@ pub async fn handle_delete_lyrics_callback_update_or_else< handle_other: HandleFn, ) -> Result where - MemberApi: DeleteLyricsMemberApi + Sync, - Effects: DeleteLyricsEffects + Sync, + MemberApi: DeleteMemberApi + Sync, + Effects: DeleteCallbackEffects + Sync, HandleFn: FnOnce(TelegramUpdate) -> HandleFuture, HandleFuture: Future>, HandleError: fmt::Display, @@ -206,15 +160,15 @@ pub async fn handle_delete_lyrics_callback( data: DeleteLyricsCallbackData, ) -> DeleteLyricsCallbackOutcome where - MemberApi: DeleteLyricsMemberApi + Sync, - Effects: DeleteLyricsEffects + Sync, + MemberApi: DeleteMemberApi + Sync, + Effects: DeleteCallbackEffects + Sync, { if data.user_id == 0 || data.chat_id == 0 { try_answer_callback(effects, &query.id, "", false, "bad lyrics data ack").await; return DeleteLyricsCallbackOutcome::BadDataAck; } - if !is_authorized_for_delete_lyrics(member_api, query, data.user_id, data.chat_id).await { + if !can_delete_owned_content(member_api, query, data.user_id, data.chat_id).await { try_answer_callback( effects, &query.id, @@ -270,7 +224,7 @@ async fn try_answer_callback( show_alert: bool, context: &'static str, ) where - Effects: DeleteLyricsEffects + Sync, + Effects: DeleteCallbackEffects + Sync, { let request = CallbackAnswerRequest { callback_query_id: query_id.to_owned(), @@ -279,7 +233,7 @@ async fn try_answer_callback( url: String::new(), cache_time: 0, }; - try_execute_delete_lyrics_method( + try_execute_delete_method( effects, build_callback_answer_method(&request).into(), context, @@ -292,10 +246,10 @@ async fn try_edit_reply_markup( request: EditReplyMarkupMessageRequest, context: &'static str, ) where - Effects: DeleteLyricsEffects + Sync, + Effects: DeleteCallbackEffects + Sync, { match build_edit_reply_markup_message_method(&request) { - Ok(method) => try_execute_delete_lyrics_method(effects, method.into(), context).await, + Ok(method) => try_execute_delete_method(effects, method.into(), context).await, Err(error) => tracing::warn!(message = %error, "failed to build delete lyrics method"), } } @@ -306,75 +260,17 @@ async fn try_delete_message( message_id: i64, context: &'static str, ) where - Effects: DeleteLyricsEffects + Sync, + Effects: DeleteCallbackEffects + Sync, { match build_delete_message_method(&DeleteMessageRequest { chat_id, message_id, }) { - Ok(method) => try_execute_delete_lyrics_method(effects, method.into(), context).await, + Ok(method) => try_execute_delete_method(effects, method.into(), context).await, Err(error) => tracing::warn!(message = %error, "failed to build delete lyrics method"), } } -async fn try_execute_delete_lyrics_method( - effects: &Effects, - method: TelegramOutboundMethod, - context: &'static str, -) where - Effects: DeleteLyricsEffects + Sync, -{ - let method_name = method.method_name(); - if let Err(error) = effects.execute_delete_lyrics_method(method).await { - tracing::warn!( - message = %error, - method = method_name, - context, - "delete lyrics callback Telegram side effect failed" - ); - } -} - -async fn is_authorized_for_delete_lyrics( - member_api: &MemberApi, - query: &TelegramCallbackQuery, - data_user_id: i64, - data_chat_id: i64, -) -> bool -where - MemberApi: DeleteLyricsMemberApi + Sync, -{ - let requester_id = i64::from(query.from.id); - if requester_id == data_user_id { - return true; - } - - match member_api.get_chat_member(data_chat_id, requester_id).await { - Ok(member) => telegram_member_can_delete_lyrics(&member), - Err(error) => { - tracing::warn!( - message = %error, - chat_id = data_chat_id, - user_id = requester_id, - "failed to authorize delete lyrics callback" - ); - false - } - } -} - -#[must_use] -pub fn telegram_member_can_delete_lyrics(member: &TelegramChatMember) -> bool { - match member { - TelegramChatMember::Creator(_) => true, - TelegramChatMember::Administrator(admin) => admin.can_delete_messages, - TelegramChatMember::Kicked(_) - | TelegramChatMember::Left(_) - | TelegramChatMember::Member { .. } - | TelegramChatMember::Restricted(_) => false, - } -} - fn delete_lyrics_callback_data_from_query( query: &TelegramCallbackQuery, ) -> Option { @@ -767,14 +663,14 @@ mod tests { } } - impl DeleteLyricsMemberApi for MemberApiStub { + impl DeleteMemberApi for MemberApiStub { type Error = io::Error; fn get_chat_member<'a>( &'a self, chat_id: i64, user_id: i64, - ) -> DeleteLyricsMemberFuture<'a, Self::Error> { + ) -> DeleteMemberFuture<'a, Self::Error> { Box::pin(async move { self.calls .lock() @@ -804,13 +700,13 @@ mod tests { } } - impl DeleteLyricsEffects for EffectsStub { + impl DeleteCallbackEffects for EffectsStub { type Error = io::Error; - fn execute_delete_lyrics_method<'a>( + fn execute_delete_method<'a>( &'a self, method: TelegramOutboundMethod, - ) -> DeleteLyricsEffectFuture<'a, Self::Error> { + ) -> DeleteEffectFuture<'a, Self::Error> { Box::pin(async move { let name = method.method_name().to_owned(); let payload = method_payload(method)?; diff --git a/crates/openplotva-app/src/delete_message.rs b/crates/openplotva-app/src/delete_message.rs index 086715c5..11319ead 100644 --- a/crates/openplotva-app/src/delete_message.rs +++ b/crates/openplotva-app/src/delete_message.rs @@ -1,6 +1,15 @@ //! App-level generic callback message deletion behavior. -use std::{fmt, future::Future, pin::Pin, sync::Arc}; +#[cfg(test)] +use crate::callbacks::DeleteEffectFuture; +#[cfg(test)] +use crate::callbacks::DeleteMemberFuture; +use crate::callbacks::{ + DeleteCallbackEffects, DeleteMemberApi, telegram_member_can_delete, try_execute_delete_method, +}; +#[cfg(test)] +use openplotva_telegram::TelegramOutboundMethod; +use std::{fmt, future::Future, sync::Arc}; use carapax::types::{ CallbackQuery as TelegramCallbackQuery, ChatMember as TelegramChatMember, @@ -8,8 +17,7 @@ use carapax::types::{ }; use openplotva_telegram::{ CallbackActionParse, CallbackAnswerRequest, ChatRef, DeleteMessageRequest, - TelegramOutboundMethod, build_callback_answer_method, build_delete_message_method, - build_get_chat_member_method, parse_callback_action, + build_callback_answer_method, build_delete_message_method, parse_callback_action, }; use thiserror::Error; @@ -18,63 +26,6 @@ use crate::updates::UpdateHandler; const DELETE_CALLBACK_ACTION: &str = "delete"; const DELETE_CALLBACK_UNAUTHORIZED_TEXT: &str = "Π― ΠΏΠΎΠ΄Ρ‡ΠΈΠ½ΡŽΡΡŒ Ρ‚ΠΎΠ»ΡŒΠΊΠΎ Π°Π²Ρ‚ΠΎΡ€Ρƒ ΠΈΠ»ΠΈ администраторам."; -/// Boxed future returned by Telegram membership APIs. -pub type DeleteMessageMemberFuture<'a, E> = - Pin> + Send + 'a>>; - -pub trait DeleteMessageMemberApi { - /// Concrete API error. - type Error: fmt::Display + Send + Sync + 'static; - - /// Fetch one Telegram chat member. - fn get_chat_member<'a>( - &'a self, - chat_id: i64, - user_id: i64, - ) -> DeleteMessageMemberFuture<'a, Self::Error>; -} - -impl DeleteMessageMemberApi for openplotva_telegram::TelegramClient { - type Error = carapax::api::ExecuteError; - - fn get_chat_member<'a>( - &'a self, - chat_id: i64, - user_id: i64, - ) -> DeleteMessageMemberFuture<'a, Self::Error> { - Box::pin(async move { - self.execute(build_get_chat_member_method(chat_id, user_id)) - .await - }) - } -} - -/// Boxed future returned by delete callback effects. -pub type DeleteMessageEffectFuture<'a, E> = - Pin> + Send + 'a>>; - -pub trait DeleteMessageEffects { - /// Concrete effect error. - type Error: fmt::Display + Send + Sync + 'static; - - /// Execute one direct Telegram Bot API method. - fn execute_delete_message_method<'a>( - &'a self, - method: TelegramOutboundMethod, - ) -> DeleteMessageEffectFuture<'a, Self::Error>; -} - -impl DeleteMessageEffects for openplotva_telegram::TelegramClient { - type Error = carapax::api::ExecuteError; - - fn execute_delete_message_method<'a>( - &'a self, - method: TelegramOutboundMethod, - ) -> DeleteMessageEffectFuture<'a, Self::Error> { - Box::pin(async move { method.execute_with(self).await.map(|_| ()) }) - } -} - #[derive(Clone, Copy, Debug, Eq, PartialEq)] pub struct DeleteMessageCallbackTarget { /// Callback message chat. @@ -132,8 +83,8 @@ impl DeleteMessageCallbackUpdateHandler UpdateHandler for DeleteMessageCallbackUpdateHandler where - MemberApi: DeleteMessageMemberApi + Send + Sync, - Effects: DeleteMessageEffects + Send + Sync, + MemberApi: DeleteMemberApi + Send + Sync, + Effects: DeleteCallbackEffects + Send + Sync, Next: UpdateHandler + Send + Sync, { type Error = DeleteMessageCallbackError; @@ -163,8 +114,8 @@ pub async fn handle_delete_message_callback_update_or_else< handle_other: HandleFn, ) -> Result where - MemberApi: DeleteMessageMemberApi + Sync, - Effects: DeleteMessageEffects + Sync, + MemberApi: DeleteMemberApi + Sync, + Effects: DeleteCallbackEffects + Sync, HandleFn: FnOnce(TelegramUpdate) -> HandleFuture, HandleFuture: Future>, HandleError: fmt::Display, @@ -196,8 +147,8 @@ pub async fn handle_delete_message_callback( query: &TelegramCallbackQuery, ) -> DeleteMessageCallbackOutcome where - MemberApi: DeleteMessageMemberApi + Sync, - Effects: DeleteMessageEffects + Sync, + MemberApi: DeleteMemberApi + Sync, + Effects: DeleteCallbackEffects + Sync, { let Some(target) = delete_message_callback_target(query) else { try_answer_callback( @@ -268,20 +219,7 @@ fn delete_callback_allows_delete( member: &TelegramChatMember, ) -> bool { target.reply_chat_id == target.message_chat.id - && (target.reply_author_user_id == Some(requester_id) - || telegram_member_can_delete_message(member)) -} - -#[must_use] -pub fn telegram_member_can_delete_message(member: &TelegramChatMember) -> bool { - match member { - TelegramChatMember::Creator(_) => true, - TelegramChatMember::Administrator(admin) => admin.can_delete_messages, - TelegramChatMember::Kicked(_) - | TelegramChatMember::Left(_) - | TelegramChatMember::Member { .. } - | TelegramChatMember::Restricted(_) => false, - } + && (target.reply_author_user_id == Some(requester_id) || telegram_member_can_delete(member)) } fn is_delete_message_callback(query: &TelegramCallbackQuery) -> bool { @@ -326,7 +264,7 @@ async fn try_answer_callback( cache_time: i64, context: &'static str, ) where - Effects: DeleteMessageEffects + Sync, + Effects: DeleteCallbackEffects + Sync, { let request = CallbackAnswerRequest { callback_query_id: query_id.to_owned(), @@ -335,7 +273,7 @@ async fn try_answer_callback( url: String::new(), cache_time, }; - try_execute_delete_message_method( + try_execute_delete_method( effects, build_callback_answer_method(&request).into(), context, @@ -349,35 +287,17 @@ async fn try_delete_message( message_id: i64, context: &'static str, ) where - Effects: DeleteMessageEffects + Sync, + Effects: DeleteCallbackEffects + Sync, { match build_delete_message_method(&DeleteMessageRequest { chat_id, message_id, }) { - Ok(method) => try_execute_delete_message_method(effects, method.into(), context).await, + Ok(method) => try_execute_delete_method(effects, method.into(), context).await, Err(error) => tracing::warn!(message = %error, "failed to build delete callback method"), } } -async fn try_execute_delete_message_method( - effects: &Effects, - method: TelegramOutboundMethod, - context: &'static str, -) where - Effects: DeleteMessageEffects + Sync, -{ - let method_name = method.method_name(); - if let Err(error) = effects.execute_delete_message_method(method).await { - tracing::warn!( - message = %error, - method = method_name, - context, - "generic delete callback Telegram side effect failed" - ); - } -} - #[cfg(test)] mod tests { use super::*; @@ -671,9 +591,9 @@ mod tests { #[test] fn delete_message_member_permission_matches_go_rights() { - assert!(telegram_member_can_delete_message(&creator(1))); - assert!(telegram_member_can_delete_message(&deleting_admin(2))); - assert!(!telegram_member_can_delete_message(&member(3))); + assert!(telegram_member_can_delete(&creator(1))); + assert!(telegram_member_can_delete(&deleting_admin(2))); + assert!(!telegram_member_can_delete(&member(3))); } #[derive(Default)] @@ -695,14 +615,14 @@ mod tests { } } - impl DeleteMessageMemberApi for MemberApiStub { + impl DeleteMemberApi for MemberApiStub { type Error = io::Error; fn get_chat_member<'a>( &'a self, chat_id: i64, user_id: i64, - ) -> DeleteMessageMemberFuture<'a, Self::Error> { + ) -> DeleteMemberFuture<'a, Self::Error> { Box::pin(async move { self.calls .lock() @@ -732,13 +652,13 @@ mod tests { } } - impl DeleteMessageEffects for EffectsStub { + impl DeleteCallbackEffects for EffectsStub { type Error = io::Error; - fn execute_delete_message_method<'a>( + fn execute_delete_method<'a>( &'a self, method: TelegramOutboundMethod, - ) -> DeleteMessageEffectFuture<'a, Self::Error> { + ) -> DeleteEffectFuture<'a, Self::Error> { Box::pin(async move { let name = method.method_name().to_owned(); let payload = method_payload(method)?; diff --git a/crates/openplotva-app/src/dialog_debounce.rs b/crates/openplotva-app/src/dialog_debounce.rs index db840aea..feb05147 100644 --- a/crates/openplotva-app/src/dialog_debounce.rs +++ b/crates/openplotva-app/src/dialog_debounce.rs @@ -17,7 +17,7 @@ use tokio::task::JoinHandle; use crate::edited::EditedDialogJobUpdate; -pub const GO_DIALOG_DEBOUNCE_INTERVAL: Duration = Duration::from_secs(5); +pub const DIALOG_DEBOUNCE_INTERVAL: Duration = Duration::from_secs(5); pub type DialogDebounceAssignObserverFuture<'a> = Pin + Send + 'a>>; pub trait DialogDebounceAssignObserver: Send + Sync { @@ -104,7 +104,7 @@ impl InMemoryDialogDebounce { #[must_use] pub fn delay_or_default(delay: Duration) -> Duration { if delay.is_zero() { - GO_DIALOG_DEBOUNCE_INTERVAL + DIALOG_DEBOUNCE_INTERVAL } else { delay } @@ -357,13 +357,13 @@ mod tests { use crate::edited::EditedDialogJobUpdate; - use super::{DialogDebounceKey, GO_DIALOG_DEBOUNCE_INTERVAL, InMemoryDialogDebounce}; + use super::{DIALOG_DEBOUNCE_INTERVAL, DialogDebounceKey, InMemoryDialogDebounce}; #[test] fn debounce_delay_uses_go_default_for_zero_only() { assert_eq!( InMemoryDialogDebounce::delay_or_default(Duration::ZERO), - GO_DIALOG_DEBOUNCE_INTERVAL + DIALOG_DEBOUNCE_INTERVAL ); assert_eq!( InMemoryDialogDebounce::delay_or_default(Duration::from_millis(150)), diff --git a/crates/openplotva-app/src/dialog_jobs/effects.rs b/crates/openplotva-app/src/dialog_jobs/effects.rs index 5bf42a61..a1df7ddb 100644 --- a/crates/openplotva-app/src/dialog_jobs/effects.rs +++ b/crates/openplotva-app/src/dialog_jobs/effects.rs @@ -15,7 +15,6 @@ use openplotva_telegram::{ build_text_message_methods_without_link_previews, build_text_message_methods_without_link_previews_with_atomic_tail, }; -use sqlx::Row; use thiserror::Error; use time::OffsetDateTime; @@ -29,15 +28,7 @@ use super::{ DialogJobEffectFuture, DialogJobReceiptFuture, DialogJobReceiptLookupFuture, dialog_response_requires_rich, }; - -/// Identity returned after all final-answer parts were accepted by one queue. -#[derive(Clone, Debug, Eq, PartialEq)] -pub struct QueuedBatchReceipt { - pub batch_id: String, - pub operation_ids: Vec, - delivery: QueuedDelivery, - delivery_complete: bool, -} +use crate::telegram_outbox::{QueuedBatchReceipt, telegram_outbox_ordering_key}; /// Final-answer delivery behavior derived from the completed dialog session. #[derive(Clone, Copy, Debug, Default, Eq, PartialEq)] @@ -56,48 +47,6 @@ impl DialogAnswerSendOptions { } } -#[derive(Clone, Copy, Debug, Eq, PartialEq)] -enum QueuedDelivery { - Dispatcher, - DurableOutbox, -} - -impl QueuedBatchReceipt { - pub(crate) fn dispatcher(batch_id: String, operation_ids: Vec) -> Self { - Self { - batch_id, - operation_ids, - delivery: QueuedDelivery::Dispatcher, - delivery_complete: false, - } - } - - pub(crate) fn durable_outbox( - batch_id: String, - operation_ids: Vec, - delivery_complete: bool, - ) -> Self { - Self { - batch_id, - operation_ids, - delivery: QueuedDelivery::DurableOutbox, - delivery_complete, - } - } - - /// Whether taskman must wait for real Telegram receipts before completion. - #[must_use] - pub fn requires_delivery_wait(&self) -> bool { - self.delivery == QueuedDelivery::DurableOutbox - } - - /// Whether every durable part already has a real Telegram receipt. - #[must_use] - pub fn delivery_complete(&self) -> bool { - self.requires_delivery_wait() && self.delivery_complete - } -} - /// Side effects performed after a provider returns a dialog answer. pub trait DialogJobEffects { /// Error returned by concrete side effects. @@ -345,30 +294,19 @@ impl DialogJobEffects for DialogDispatcherEffects { return Ok(None); }; let batch_id = dialog_answer_batch_id(outbox.bot_id, dialog_job_id); - let rows = sqlx::query( - "SELECT operation_id, state FROM telegram_outbox \ - WHERE batch_id = $1 AND dialog_job_id = $2 \ - ORDER BY part_index", - ) - .bind(&batch_id) - .bind(dialog_job_id) - .fetch_all(outbox.store.pool()) - .await - .map_err(openplotva_storage::StorageError::from)?; - if rows.is_empty() { + let Some(queued) = outbox + .store + .queued_batch(outbox.bot_id, &batch_id, Some(dialog_job_id)) + .await? + else { return Ok(None); - } - let delivery_complete = rows.iter().all(|row| { - matches!( - row.try_get::("state").as_deref(), - Ok("delivered") - ) - }); - let operation_ids = rows + }; + let delivery_complete = queued.parts.iter().all(|part| part.state == "delivered"); + let operation_ids = queued + .parts .into_iter() - .map(|row| row.try_get("operation_id")) - .collect::, sqlx::Error>>() - .map_err(openplotva_storage::StorageError::from)?; + .map(|part| part.operation_id) + .collect(); Ok(Some(QueuedBatchReceipt::durable_outbox( batch_id, operation_ids, @@ -562,6 +500,7 @@ fn dialog_intermediate_outbox_batch( parts: Vec, ) -> TelegramOutboxBatchInput { TelegramOutboxBatchInput { + content_policy: openplotva_storage::TelegramOutboxContentPolicy::Immutable, batch_id: if params.meta["agent_final_reply"].as_bool() == Some(true) { dialog_answer_batch_id(bot_id, dialog_job_id) } else { @@ -570,7 +509,7 @@ fn dialog_intermediate_outbox_batch( bot_id, chat_id: Some(params.chat_id), thread_id: params.thread_id, - ordering_key: dialog_outbox_ordering_key(bot_id, params.chat_id, params.thread_id), + ordering_key: telegram_outbox_ordering_key(bot_id, params.chat_id, params.thread_id), causation_update_id, dialog_job_id: Some(dialog_job_id), trigger_message_id: Some(i64::from(params.message_id)), @@ -582,17 +521,6 @@ fn dialog_intermediate_outbox_batch( } /// Per-chat delivery order shared by dialog messages and the banners queued after them. -pub(crate) fn dialog_outbox_ordering_key( - bot_id: i64, - chat_id: i64, - thread_id: Option, -) -> String { - format!( - "dialog:{bot_id}:{chat_id}:{}", - thread_id.unwrap_or_default() - ) -} - fn dialog_answer_batch_id(bot_id: i64, dialog_job_id: i64) -> String { format!("dialog-answer:v1:{bot_id}:{dialog_job_id}") } @@ -662,11 +590,12 @@ async fn enqueue_durable_dialog_answer( .collect::, _>>()?; let batch_id = dialog_answer_batch_id(outbox.bot_id, dialog_job_id); let batch = TelegramOutboxBatchInput { + content_policy: openplotva_storage::TelegramOutboxContentPolicy::Immutable, batch_id: batch_id.clone(), bot_id: outbox.bot_id, chat_id: Some(params.chat_id), thread_id: params.thread_id, - ordering_key: dialog_outbox_ordering_key(outbox.bot_id, params.chat_id, params.thread_id), + ordering_key: telegram_outbox_ordering_key(outbox.bot_id, params.chat_id, params.thread_id), causation_update_id, dialog_job_id: Some(dialog_job_id), trigger_message_id: Some(i64::from(params.message_id)), diff --git a/crates/openplotva-app/src/dialog_jobs/mod.rs b/crates/openplotva-app/src/dialog_jobs/mod.rs index 9a80ce56..8fbd9e94 100644 --- a/crates/openplotva-app/src/dialog_jobs/mod.rs +++ b/crates/openplotva-app/src/dialog_jobs/mod.rs @@ -5,6 +5,7 @@ use std::{future::Future, pin::Pin, time::Duration}; +use crate::telegram_outbox::QueuedBatchReceipt; use openplotva_dialog::DialogInput; use openplotva_taskman::{DIALOG_AIFARM_QUEUE_NAME, TEXT_QUEUE_NAME}; use thiserror::Error; diff --git a/crates/openplotva-app/src/dialog_messages.rs b/crates/openplotva-app/src/dialog_messages.rs index b19d8c56..5908aef7 100644 --- a/crates/openplotva-app/src/dialog_messages.rs +++ b/crates/openplotva-app/src/dialog_messages.rs @@ -2277,7 +2277,7 @@ where .ok() .flatten() .unwrap_or_else(|| ChatSettings::defaults(chat_id)); - if settings_row.reactivity_percentage == GO_SETTINGS_REACTIVITY_OFF { + if settings_row.reactivity_percentage == SETTINGS_REACTIVITY_OFF { return Ok(DialogMessageUpdateRoute::RandomSkippedReactivityOff); } @@ -2409,7 +2409,7 @@ fn aifarm_dialog_max_output_tokens( } fn random_response_triggers(reactivity_percentage: i32, roll: i32) -> bool { - if reactivity_percentage == GO_SETTINGS_REACTIVITY_OFF { + if reactivity_percentage == SETTINGS_REACTIVITY_OFF { return false; } roll > 100 - reactivity_percentage @@ -2443,7 +2443,7 @@ fn obscenify_string(text: &str, variant_roll: i32) -> String { fn first_obscenify_vowel_with_suffix(value: &str) -> Option<(&str, &str)> { let (index, _) = value .char_indices() - .find(|(_, ch)| GO_OBSCENIFY_VOWELS.contains(*ch))?; + .find(|(_, ch)| OBSCENIFY_VOWELS.contains(*ch))?; let next = index + value[index..].chars().next()?.len_utf8(); (next < value.len()).then_some((&value[index..next], &value[next..])) } @@ -2530,8 +2530,8 @@ const LONG_DIALOG_PHRASES: &[&str] = &[ "review", ]; -const GO_SETTINGS_REACTIVITY_OFF: i32 = 0; -const GO_OBSCENIFY_VOWELS: &str = "Π°ΡƒΠΎΡ‹ΠΈΡΡΡŽΡ‘Π΅"; +const SETTINGS_REACTIVITY_OFF: i32 = 0; +const OBSCENIFY_VOWELS: &str = "Π°ΡƒΠΎΡ‹ΠΈΡΡΡŽΡ‘Π΅"; #[cfg(test)] mod tests { @@ -2657,7 +2657,7 @@ mod tests { let scheduler = TaskmanDialogMessageScheduler::new( Arc::clone(&queue), debounce, - crate::dialog_debounce::GO_DIALOG_DEBOUNCE_INTERVAL, + crate::dialog_debounce::DIALOG_DEBOUNCE_INTERVAL, durability, ) .with_typing_effects(typing); @@ -2700,7 +2700,7 @@ mod tests { let scheduler = TaskmanDialogMessageScheduler::new( Arc::clone(&queue), Arc::new(InMemoryDialogDebounce::new()), - crate::dialog_debounce::GO_DIALOG_DEBOUNCE_INTERVAL, + crate::dialog_debounce::DIALOG_DEBOUNCE_INTERVAL, Arc::new(NoopDialogTaskDurabilityBarrier), ) .with_typing_effects(Arc::new(PendingTypingStub)); @@ -5520,6 +5520,20 @@ mod tests { let Ok(redis_url) = env::var("OPENPLOTVA_TEST_REDIS_URL") else { return Ok(()); }; + let Ok(dsn) = env::var("OPENPLOTVA_TEST_POSTGRES_DSN") else { + return Ok(()); + }; + let pool = sqlx::postgres::PgPoolOptions::new() + .max_connections(4) + .connect(&dsn) + .await?; + openplotva_storage::run_migrations_on(&pool).await?; + let (runtime, _) = + crate::task_queue::SharedTaskQueueRuntime::load_from_postgres_with_id_allocator( + openplotva_storage::PostgresTaskQueueStore::new(pool.clone()), + openplotva_taskman::TaskQueueIdAllocator::default(), + ) + .await?; let redis_client = redis::Client::open(redis_url)?; let suffix = SystemTime::now().duration_since(UNIX_EPOCH)?.as_nanos(); let key = format!("openplotva:test:decoded-song:{suffix}"); @@ -5542,7 +5556,7 @@ mod tests { .ok_or_else(|| std::io::Error::other("expected decoded song update"))?; let state_store = UpdateStateStoreStub::default(); - let queue = Arc::new(InMemoryTaskQueue::new()); + let queue = runtime.queue(); let song_scheduler = crate::dialog_tools::TaskmanDialogToolAdapter::new(Arc::clone(&queue)) .with_draw_image_vip_status(Arc::new(VipStatusStub(true))); let scheduler = SchedulerStub::default(); @@ -5602,19 +5616,34 @@ mod tests { "user:99:Ada:ada_l".to_owned() ] ); - assert_eq!( + assert!(matches!( *lock(&route), - Some(DialogMessageUpdateRoute::SongScheduled { - status: "scheduled".to_owned(), - no_reply: false, - }) - ); - assert!(scheduler.calls().is_empty()); + Some(DialogMessageUpdateRoute::Scheduled { .. }) + )); + assert_eq!(scheduler.calls().len(), 1); + assert_eq!(scheduler.metas()[0].requested_tool, "generate_song"); assert!(effects.sent_song_notices().is_empty()); - let job_id = { + let scheduled = song_scheduler + .schedule_song(crate::dialog_tools::SongScheduleRequest { + chat_id: -100, + message_id: 78, + user_id: 99, + user_full_name: "Ada".into(), + topic: "neon rain".into(), + message_text: "!song neon rain".into(), + message_meta: scheduler.metas()[0].clone(), + ..Default::default() + }) + .await + .map_err(|error| std::io::Error::other(error.to_string()))?; + assert_eq!(scheduled.status, "scheduled"); + let job_id = scheduled.job_id.expect("scheduled song job"); + { let records = queue.records(); - assert_eq!(records.len(), 1); - let record = &records[0]; + let record = records + .iter() + .find(|record| record.id == job_id) + .expect("music job"); assert_eq!(record.queue_name, MUSIC_VIP_QUEUE_NAME); assert_eq!(record.status, JobStatus::Pending); assert_eq!(record.job.title, "music"); @@ -5626,8 +5655,7 @@ mod tests { let music = record.job.data.music_data.as_ref().expect("music"); assert_eq!(music.topic, "neon rain"); assert_eq!(music.reference_file_id, ""); - record.id - }; + } let material_provider = MusicMaterialProviderStub::new(crate::music_jobs::SongMaterial { title: "Neon Rain".to_owned(), @@ -5648,12 +5676,34 @@ mod tests { )); let music_files = MusicReferenceFileStoreCapture::default(); let music_sender = MusicTelegramSenderCapture::new(vec![]); - let music_rich = Arc::new(crate::rich::MockRichSender::default()); + let uploads = Arc::new(Mutex::new(Vec::new())); + let captured_uploads = uploads.clone(); + let listener = tokio::net::TcpListener::bind("127.0.0.1:0").await?; + let uploader = openplotva_media::uploader::UploaderClient::new( + openplotva_media::uploader::UploaderConfig { + base_url: format!("http://{}", listener.local_addr()?), + secret: "test-secret".into(), + ..Default::default() + }, + )?; + let server = + tokio::spawn(async move { + axum::serve(listener, axum::Router::new().route("/upload", axum::routing::post( + move |headers: axum::http::HeaderMap, body: axum::body::Bytes| async move { + assert_eq!(headers["authorization"], "Bearer test-secret"); + captured_uploads.lock().expect("uploads").push(body); + axum::Json(serde_json::json!({"url":"https://media.example.test/song.mp3"})) + } + ))).await + }); + let store = openplotva_storage::PostgresTelegramOutboxStore::new(pool.clone()); let music_effects = crate::music_jobs::TelegramMusicJobEffects::new( permissions, music_files.clone(), music_sender.clone(), - music_rich.clone(), + uploader, + Arc::new(crate::rich::RichOutbox::new(store.clone(), 7)), + openplotva_storage::PostgresGeneratedSongStore::new(pool.clone()), ); let worker_report = crate::music_jobs::run_music_queue_once( &queue, @@ -5669,7 +5719,7 @@ mod tests { crate::music_jobs::MusicQueuePollReport { queue_name: MUSIC_VIP_QUEUE_NAME.to_owned(), job_id: Some(job_id), - outcome: crate::music_jobs::MusicQueuePollOutcome::Completed, + outcome: crate::music_jobs::MusicQueuePollOutcome::WaitingDelivery, error: None, activity: crate::telegram_activity::TelegramActivitySnapshot::default(), } @@ -5700,17 +5750,127 @@ mod tests { music_sender.methods().is_empty(), "song now sends one rich message, not classic audio + lyrics" ); - { - let sent = music_rich.sent.lock().expect("rich sent"); - assert_eq!(sent.len(), 1); - assert_eq!(sent[0].chat_id, -100); - assert_eq!(sent[0].reply_to_message_id, Some(78)); - assert!(sent[0].html.contains(r#"