diff --git a/.env.example b/.env.example index 9831085a..9bc26567 100644 --- a/.env.example +++ b/.env.example @@ -153,6 +153,8 @@ # Retention (days) for the whitecircle_checks audit log; 0 disables the cleanup worker. # WHITECIRCLE_CHECKS_RETENTION_DAYS=30 # Privacy-preserving ads (server-side only; disabled by default). +# Advertising is restricted to the initiating user's private chat with the bot. +# These switches never enable advertising in groups or channels. # GRADIUS_ENABLED=false # GRADIUS_UTILITY_IMAGE_ENABLED=false # GRADIUS_UTILITY_RATES_ENABLED=false diff --git a/Cargo.lock b/Cargo.lock index a54cc5d8..74ae443e 100644 --- a/Cargo.lock +++ b/Cargo.lock @@ -497,9 +497,6 @@ name = "bitflags" version = "2.11.1" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "c4512299f36f043ab09a583e57bceb5a5aab7a73db1805848e8fef3c9e8c78b3" -dependencies = [ - "serde_core", -] [[package]] name = "block-buffer" @@ -1131,17 +1128,6 @@ dependencies = [ "miniz_oxide", ] -[[package]] -name = "flume" -version = "0.12.0" -source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "5e139bc46ca777eb5efaf62df0ab8cc5fd400866427e56c68b22e414e53bd3be" -dependencies = [ - "futures-core", - "futures-sink", - "spin", -] - [[package]] name = "fnv" version = "1.0.7" @@ -1191,17 +1177,6 @@ version = "0.3.32" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "7e3450815272ef58cec6d564423f6e755e25379b217b0bc688e295ba24df6b1d" -[[package]] -name = "futures-executor" -version = "0.3.32" -source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "baf29c38818342a3b26b5b923639e7b1f4a61fc5e76102d4b1981c6dc7a7579d" -dependencies = [ - "futures-core", - "futures-task", - "futures-util", -] - [[package]] name = "futures-intrusive" version = "0.5.0" @@ -1832,16 +1807,6 @@ version = "0.2.186" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "68ab91017fe16c622486840e4c83c9a37afeff978bd239b5293d61ece587de66" -[[package]] -name = "libsqlite3-sys" -version = "0.35.0" -source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "133c182a6a2c87864fe97778797e46c7e999672690dc9fa3ee8e241aa4a9c13f" -dependencies = [ - "pkg-config", - "vcpkg", -] - [[package]] name = "linux-raw-sys" version = "0.12.1" @@ -2134,9 +2099,8 @@ dependencies = [ "reqwest", "serde", "serde_json", - "sha1 0.10.6", + "sha1", "sha2 0.10.9", - "sqlx", "thiserror", "time", "tokio", @@ -2328,7 +2292,8 @@ dependencies = [ "serde", "serde_json", "sha2 0.10.9", - "sqlx", + "sqlx-core", + "sqlx-postgres", "thiserror", "time", "tokio", @@ -2359,7 +2324,7 @@ dependencies = [ "reqwest", "serde", "serde_json", - "sha1 0.10.6", + "sha1", "thiserror", "time", "tokio", @@ -3200,17 +3165,6 @@ dependencies = [ "digest 0.10.7", ] -[[package]] -name = "sha1" -version = "0.11.0" -source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "aacc4cc499359472b4abe1bf11d0b12e688af9a805fa5e3016f9a386dc2d0214" -dependencies = [ - "cfg-if", - "cpufeatures 0.3.0", - "digest 0.11.3", -] - [[package]] name = "sha1_smol" version = "1.0.1" @@ -3332,22 +3286,6 @@ name = "spin" version = "0.9.8" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "6980e8d7511241f8acf4aebddbb1ff938df5eebe98691418c4468d0b72a96a67" -dependencies = [ - "lock_api", -] - -[[package]] -name = "sqlx" -version = "0.9.0" -source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "378620ccc25c62c89d8be1c819e76a88d59bdcc3304733330788948e619bfd71" -dependencies = [ - "sqlx-core", - "sqlx-macros", - "sqlx-mysql", - "sqlx-postgres", - "sqlx-sqlite", -] [[package]] name = "sqlx-core" @@ -3384,72 +3322,6 @@ dependencies = [ "url", ] -[[package]] -name = "sqlx-macros" -version = "0.9.0" -source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "bd2b84f2bc39a5705ef27ec785a11c934a41bbd4a24941e257927cddc26b60bf" -dependencies = [ - "proc-macro2", - "quote", - "sqlx-core", - "sqlx-macros-core", - "syn 2.0.117", -] - -[[package]] -name = "sqlx-macros-core" -version = "0.9.0" -source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "fb8d96de5fdc85a5c4ec813432b523ec637e80ba98f046555f75f7908ddac7c3" -dependencies = [ - "cfg-if", - "dotenvy", - "either", - "heck", - "hex", - "proc-macro2", - "quote", - "serde", - "serde_json", - "sha2 0.10.9", - "sqlx-core", - "sqlx-mysql", - "sqlx-postgres", - "sqlx-sqlite", - "syn 2.0.117", - "thiserror", - "tokio", - "url", -] - -[[package]] -name = "sqlx-mysql" -version = "0.9.0" -source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "90b8020fe17c5f2c245bfa2505d7ef59c5604839527c740266ad2214acebea27" -dependencies = [ - "bitflags", - "byteorder", - "bytes", - "crc", - "digest 0.11.3", - "dotenvy", - "either", - "futures-core", - "futures-util", - "generic-array", - "log", - "percent-encoding", - "serde", - "sha1 0.11.0", - "sha2 0.11.0", - "sqlx-core", - "thiserror", - "time", - "tracing", -] - [[package]] name = "sqlx-postgres" version = "0.9.0" @@ -3486,31 +3358,6 @@ dependencies = [ "whoami", ] -[[package]] -name = "sqlx-sqlite" -version = "0.9.0" -source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "488e99c397a62007e4229aec669a179816339afc6d2620ca6fa420dbee2e982c" -dependencies = [ - "atoi", - "flume", - "form_urlencoded", - "futures-channel", - "futures-core", - "futures-executor", - "futures-intrusive", - "futures-util", - "libsqlite3-sys", - "log", - "percent-encoding", - "serde", - "sqlx-core", - "thiserror", - "time", - "tracing", - "url", -] - [[package]] name = "stable_deref_trait" version = "1.2.1" @@ -4090,12 +3937,6 @@ version = "0.1.1" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "ba73ea9cf16a25df0c8caa16c51acb937d5712a8429db78a3ee29d5dcacd3a65" -[[package]] -name = "vcpkg" -version = "0.2.15" -source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "accd4ea62f7bb7a82fe23066fb0957d48ef677f6eeb8215f372f52e48bb32426" - [[package]] name = "version_check" version = "0.9.5" diff --git a/Cargo.toml b/Cargo.toml index c2f4bc36..6f681d50 100644 --- a/Cargo.toml +++ b/Cargo.toml @@ -63,7 +63,8 @@ serde = "1.0.228" serde_json = "1.0" sha1 = "0.10.6" sha2 = "0.10.9" -sqlx = { version = "0.9.0", default-features = false } +sqlx-core = { version = "0.9.0", default-features = false } +sqlx-postgres = { version = "0.9.0", default-features = false } thiserror = "2.0.18" time = "0.3" tokio = "1.52.3" diff --git a/crates/openplotva-app/Cargo.toml b/crates/openplotva-app/Cargo.toml index a20446f5..9c31e7f3 100644 --- a/crates/openplotva-app/Cargo.toml +++ b/crates/openplotva-app/Cargo.toml @@ -41,7 +41,6 @@ serde = { workspace = true, features = ["derive"] } serde_json.workspace = true sha1.workspace = true sha2.workspace = true -sqlx = { workspace = true, features = ["json", "postgres", "runtime-tokio", "time"] } thiserror.workspace = true time = { workspace = true, features = ["formatting", "parsing"] } tokio = { workspace = true, features = ["macros", "net", "rt-multi-thread", "signal", "sync", "time"] } diff --git a/crates/openplotva-app/src/checkin.rs b/crates/openplotva-app/src/checkin.rs index aad467f2..f5cacea2 100644 --- a/crates/openplotva-app/src/checkin.rs +++ b/crates/openplotva-app/src/checkin.rs @@ -2766,6 +2766,7 @@ mod tests { TelegramFileMetadataStoreFuture, UpdateHandler, UpdateStateStore, UpdateStateStoreFuture, process_update_with_state_store_at, }; + use openplotva_storage::database as sqlx; use openplotva_taskman::{JobPayload, JobType, control_job_params_from_stateless_job}; use openplotva_telegram::{ CallbackHandlerKind, DispatcherConfig, DispatcherQueue, TelegramOutboundMethodKind, diff --git a/crates/openplotva-app/src/dialog_messages.rs b/crates/openplotva-app/src/dialog_messages.rs index 5908aef7..3635b399 100644 --- a/crates/openplotva-app/src/dialog_messages.rs +++ b/crates/openplotva-app/src/dialog_messages.rs @@ -2536,6 +2536,7 @@ const OBSCENIFY_VOWELS: &str = "ауоыиэяюёе"; #[cfg(test)] mod tests { use crate::dialog_tools::DrawImageScheduleRequest; + use openplotva_storage::database as sqlx; use openplotva_telegram::TELEGRAM_PARSE_MODE_HTML; use std::{ collections::VecDeque, diff --git a/crates/openplotva-app/src/dialog_turn/ledger.rs b/crates/openplotva-app/src/dialog_turn/ledger.rs index 567a390b..832fd6f1 100644 --- a/crates/openplotva-app/src/dialog_turn/ledger.rs +++ b/crates/openplotva-app/src/dialog_turn/ledger.rs @@ -2,6 +2,7 @@ //! `RuntimeLlmObserver` pattern in `runtime_llm.rs`. One record per dialog //! worker tick that dequeued a job. +use openplotva_storage::database as sqlx; use std::sync::{ Arc, Mutex, atomic::{AtomicI64, Ordering}, @@ -443,6 +444,7 @@ pub async fn delete_old_turn_outcomes_batch( #[cfg(test)] mod tests { use super::*; + use openplotva_storage::database as sqlx; use sqlx::Execute; fn sample_record(job_id: i64, outcome: &str) -> DialogTurnOutcomeRecord { diff --git a/crates/openplotva-app/src/gradius_ads.rs b/crates/openplotva-app/src/gradius_ads.rs index e91122b3..a5967d22 100644 --- a/crates/openplotva-app/src/gradius_ads.rs +++ b/crates/openplotva-app/src/gradius_ads.rs @@ -1,3 +1,4 @@ +use openplotva_storage::database as sqlx; use std::{future::Future, pin::Pin, sync::Arc}; use openplotva_llm::gradius::vip_hint_for_impression; @@ -542,6 +543,10 @@ pub struct GradiusUtilityAdRequest { pub completed_at: OffsetDateTime, } +pub(crate) const fn is_private_ad_chat(chat_id: i64, user_id: i64) -> bool { + chat_id > 0 && chat_id == user_id +} + pub(crate) fn initialize_placement_service( config: &openplotva_config::AppConfig, api_key: &str, @@ -595,17 +600,13 @@ impl GradiusUtilityAdService { &self, request: GradiusUtilityAdRequest, ) -> Result, String> { - if request.user_id <= 0 + if !is_private_ad_chat(request.chat_id, request.user_id) || request.source_id.is_empty() || request.source_id.len() > 128 || !request .source_id .bytes() .all(|byte| byte.is_ascii_alphanumeric() || matches!(byte, b':' | b'-' | b'_')) - || request.chat_id == 0 - || (request.surface == GradiusUtilitySurface::Image - && request.chat_id > 0 - && request.chat_id != request.user_id) || (request.surface == GradiusUtilitySurface::Image && request .prompt @@ -976,7 +977,7 @@ impl GradiusAdService { &self, request: GradiusAdAppendRequest, ) -> Result, String> { - if request.chat_id <= 0 + if !is_private_ad_chat(request.chat_id, request.user_id) || self .vip .verified_is_vip(request.user_id, request.completed_at) @@ -1190,7 +1191,7 @@ impl GradiusAdService { &self, request: GradiusAdAppendRequest, ) -> Result<(), String> { - if request.chat_id <= 0 + if !is_private_ad_chat(request.chat_id, request.user_id) || self .vip .verified_is_vip(request.user_id, request.completed_at) @@ -1514,6 +1515,9 @@ fn banner_outbox_batch( banner: &GradiusAdBanner, target: GradiusBannerTarget, ) -> Result { + if target.chat_id <= 0 { + return Err("Advertising is restricted to private chats".to_owned()); + } let chat = ChatRef { direct_messages_topic_id: None, id: target.chat_id, @@ -1664,6 +1668,7 @@ fn is_word_character(character: char) -> bool { #[cfg(test)] mod tests { + use openplotva_storage::database as sqlx; use std::{ collections::VecDeque, future::Future, @@ -2224,9 +2229,21 @@ mod tests { assert_eq!(kind, "sendRichMessage"); let html = payload["html"].as_str().expect("rich content"); assert!(html.contains("")); - assert!(html.contains("
")); - assert!(html.find("
") < html.find("
")); - assert!(html.find("
") < html.find("📢")); + if chat_id > 0 { + assert!(html.contains("
")); + assert!(html.find("") < html.find("
")); + assert!(html.find("
") < html.find("📢")); + } else { + assert!(!html.contains("📢")); + assert!(!html.contains("
")); + let ad_count: i64 = sqlx::query_scalar( + "SELECT count(*) FROM gradius_ad_opportunities WHERE chat_id = $1", + ) + .bind(chat_id) + .fetch_one(&pool) + .await?; + assert_eq!(ad_count, 0); + } assert!(payload.get("link_preview_options").is_none()); assert_eq!(payload["options"]["reply_to_message_id"], 77); sqlx::query("DELETE FROM telegram_outbox WHERE batch_id = $1") @@ -2238,12 +2255,125 @@ mod tests { .execute(&pool) .await?; } - assert_eq!(client.calls.lock().expect("provider calls").len(), 2); + assert_eq!(client.calls.lock().expect("provider calls").len(), 1); Ok(()) } #[tokio::test] - async fn public_image_ad_has_no_vip_appendix_in_persisted_html() { + async fn all_utility_surfaces_allow_the_initiators_private_chat() { + for surface in [ + GradiusUtilitySurface::Image, + GradiusUtilitySurface::Rates, + GradiusUtilitySurface::Checkin, + ] { + let ledger = LedgerStub::default(); + let service = GradiusUtilityAdService::new( + Arc::new(UtilityStub::default()), + Arc::new(RedactorStub::default()), + Arc::new(ledger.clone()), + Arc::new(VipStub(false)), + ); + assert!( + service + .prepare(GradiusUtilityAdRequest { + surface, + include_vip_appendix: true, + source_id: "private-only".to_owned(), + attempt_key: "claim-1".to_owned(), + user_id: 42, + chat_id: 42, + thread_id: None, + prompt: Some("Portrait".to_owned()), + result_context: None, + completed_at: OffsetDateTime::UNIX_EPOCH, + }) + .await + .expect("private placement") + .is_some() + ); + assert_eq!( + ledger.reserved_inputs.lock().expect("reservations").len(), + 1 + ); + } + } + + #[tokio::test] + async fn all_utility_surfaces_skip_non_private_chats_before_external_calls() { + for surface in [ + GradiusUtilitySurface::Image, + GradiusUtilitySurface::Rates, + GradiusUtilitySurface::Checkin, + ] { + for chat_id in [-10042, 0, 43] { + let client = UtilityStub::default(); + let ledger = LedgerStub::default(); + let service = GradiusUtilityAdService::new( + Arc::new(client.clone()), + Arc::new(FailingRedactor), + Arc::new(ledger.clone()), + Arc::new(FailingVipStub), + ); + let ad = service + .prepare(GradiusUtilityAdRequest { + surface, + include_vip_appendix: true, + source_id: "private-only".to_owned(), + attempt_key: "claim-1".to_owned(), + user_id: 42, + chat_id, + thread_id: None, + prompt: Some("Portrait".to_owned()), + result_context: None, + completed_at: OffsetDateTime::UNIX_EPOCH, + }) + .await + .expect("non-private chat skips VIP and privacy lookups"); + assert!(ad.is_none()); + assert!( + ledger + .reserved_inputs + .lock() + .expect("reservations") + .is_empty() + ); + assert!(client.calls.lock().expect("utility calls").is_empty()); + assert!( + client + .generation_calls + .lock() + .expect("generation calls") + .is_empty() + ); + } + } + } + + #[test] + fn banner_delivery_rejects_non_private_targets() { + let banner = super::GradiusAdBanner { + opportunity_id: 1, + html: "

Offer

Реклама
".to_owned(), + keyboard: Vec::new(), + }; + for chat_id in [-10042, 0] { + assert!( + super::banner_outbox_batch( + 7, + &banner, + GradiusBannerTarget { + dialog_job_id: 1, + chat_id, + thread_id: None + }, + ) + .is_err() + ); + } + } + + #[tokio::test] + async fn private_image_ad_can_omit_vip_appendix_in_persisted_html() { let ledger = LedgerStub::default(); let service = GradiusUtilityAdService::new( Arc::new(UtilityStub::default()), @@ -2255,11 +2385,11 @@ mod tests { .prepare(GradiusUtilityAdRequest { surface: GradiusUtilitySurface::Image, include_vip_appendix: false, - source_id: "public-image".to_owned(), + source_id: "private-image".to_owned(), attempt_key: "claim-1".to_owned(), user_id: 100, - chat_id: -100, - thread_id: Some(12), + chat_id: 100, + thread_id: None, prompt: Some("Portrait".to_owned()), result_context: None, completed_at: OffsetDateTime::UNIX_EPOCH, @@ -2276,40 +2406,38 @@ mod tests { } #[tokio::test] - async fn private_and_ephemeral_image_ads_keep_vip_appendix_in_persisted_html() { - for chat_id in [100, -100] { - let ledger = LedgerStub::default(); - let service = GradiusUtilityAdService::new( - Arc::new(UtilityStub::default()), - Arc::new(RedactorStub::default()), - Arc::new(ledger.clone()), - Arc::new(VipStub(false)), - ); - let ad = service - .prepare(GradiusUtilityAdRequest { - surface: GradiusUtilitySurface::Image, - include_vip_appendix: true, - source_id: "personal-image".to_owned(), - attempt_key: "claim-1".to_owned(), - user_id: 100, - chat_id, - thread_id: None, - prompt: Some("Portrait".to_owned()), - result_context: None, - completed_at: OffsetDateTime::UNIX_EPOCH, - }) - .await - .expect("prepared") - .expect("ad"); - assert!( - ad.html - .starts_with("📢 Ad here\n\n") - ); - assert!(ad.html.contains("")); - assert!(ad.html.contains("https://t.me/PlotvoBot?start=vip")); - let stored = ledger.finished_ads.lock().expect("finished ads"); - assert_eq!(stored[0].1.rendered_html, ad.html); - } + async fn private_image_ads_keep_vip_appendix_in_persisted_html() { + let ledger = LedgerStub::default(); + let service = GradiusUtilityAdService::new( + Arc::new(UtilityStub::default()), + Arc::new(RedactorStub::default()), + Arc::new(ledger.clone()), + Arc::new(VipStub(false)), + ); + let ad = service + .prepare(GradiusUtilityAdRequest { + surface: GradiusUtilitySurface::Image, + include_vip_appendix: true, + source_id: "personal-image".to_owned(), + attempt_key: "claim-1".to_owned(), + user_id: 100, + chat_id: 100, + thread_id: None, + prompt: Some("Portrait".to_owned()), + result_context: None, + completed_at: OffsetDateTime::UNIX_EPOCH, + }) + .await + .expect("prepared") + .expect("ad"); + assert!( + ad.html + .starts_with("📢 Ad here\n\n") + ); + assert!(ad.html.contains("")); + assert!(ad.html.contains("https://t.me/PlotvoBot?start=vip")); + let stored = ledger.finished_ads.lock().expect("finished ads"); + assert_eq!(stored[0].1.rendered_html, ad.html); } #[tokio::test] @@ -2364,16 +2492,16 @@ mod tests { } #[tokio::test] - async fn generation_group_ads_use_initiator_and_skip_vip_or_unknown_users() { + async fn generation_private_ads_use_initiator_and_skip_vip_or_unknown_users() { let client = UtilityStub::default(); let ledger = LedgerStub::default(); let request = |user_id| GradiusUtilityAdRequest { surface: GradiusUtilitySurface::Image, include_vip_appendix: true, - source_id: format!("group-{user_id}"), + source_id: format!("private-{user_id}"), attempt_key: "claim-1".to_owned(), user_id, - chat_id: -100, + chat_id: user_id, thread_id: Some(12), prompt: Some("Portrait".to_owned()), result_context: None, @@ -2396,14 +2524,14 @@ mod tests { service .prepare(request(100)) .await - .expect("free group user") + .expect("free private user") .is_some() ); assert!( service .prepare(request(101)) .await - .expect("another free group user") + .expect("another free private user") .is_some() ); let vip = GradiusUtilityAdService::new( @@ -2420,12 +2548,12 @@ mod tests { ); let calls = client.generation_calls.lock().expect("generation calls"); assert_eq!(calls.len(), 2); - assert_eq!(calls[0].chat_id, calls[1].chat_id); + assert_ne!(calls[0].chat_id, calls[1].chat_id); assert_ne!(calls[0].user_id, calls[1].user_id); let opportunities = ledger.reserved_inputs.lock().expect("opportunities"); assert_eq!(opportunities[0].integration_kind, "native_generation"); assert_eq!(opportunities[0].user_id, 100); - assert_eq!(opportunities[0].chat_id, -100); + assert_eq!(opportunities[0].chat_id, 100); assert_eq!(opportunities[0].thread_id, 12); } @@ -2439,7 +2567,7 @@ mod tests { source_id: "privacy-test".to_owned(), attempt_key: "claim-1".to_owned(), user_id: 100, - chat_id: -100, + chat_id: 100, thread_id: None, prompt: Some("Portrait of Alice@example.com".to_owned()), result_context: None, @@ -2503,10 +2631,10 @@ mod tests { let request = |user_id| GradiusUtilityAdRequest { surface: GradiusUtilitySurface::Rates, include_vip_appendix: false, - source_id: format!("-100:{user_id}"), + source_id: format!("private:{user_id}"), attempt_key: "rates-claim".to_owned(), user_id, - chat_id: -100, + chat_id: user_id, thread_id: None, prompt: None, result_context: Some("Alice's rate table".to_owned()), @@ -2553,9 +2681,17 @@ mod tests { Arc::new(ledger.clone()), Arc::new(VipStub(false)), ); - let mut group_request = test_request(70); - group_request.chat_id = -10042; - assert_eq!(group_service.append(group_request).await, Ok(None)); + for chat_id in [-10042, 0, 43] { + let mut group_request = test_request(70); + group_request.chat_id = chat_id; + assert_eq!(group_service.append(group_request.clone()).await, Ok(None)); + assert_eq!( + group_service + .record_unsupported_surface(group_request) + .await, + Ok(()) + ); + } let vip_service = GradiusAdService::new( Arc::new(dialogue.clone()), diff --git a/crates/openplotva-app/src/gradius_utility_outbox.rs b/crates/openplotva-app/src/gradius_utility_outbox.rs index 9667f65e..509a4369 100644 --- a/crates/openplotva-app/src/gradius_utility_outbox.rs +++ b/crates/openplotva-app/src/gradius_utility_outbox.rs @@ -15,14 +15,14 @@ use openplotva_telegram::{ }; use time::OffsetDateTime; -use crate::gradius_ads::{GradiusUtilityAdRequest, GradiusUtilityAdService, GradiusUtilitySurface}; +use crate::gradius_ads::{ + GradiusUtilityAdRequest, GradiusUtilityAdService, GradiusUtilitySurface, is_private_ad_chat, +}; #[derive(Clone)] pub struct GradiusUtilityImageAds { pub ads: Arc, pub outbox: Arc, - pub telegram: openplotva_telegram::TelegramClient, - pub bot_id: i64, } impl std::fmt::Debug for GradiusUtilityImageAds { @@ -45,24 +45,16 @@ impl GradiusUtilityImageAds { prompt: String, first_photo_id: i32, ) -> Result<(), String> { - let delivery = image_ad_delivery(&self.telegram, self.bot_id, chat_id).await?; - if delivery == ImageAdDelivery::Persistent - && chat_id < 0 - && !self - .outbox - .store - .public_image_group_available(chat_id) - .await - .map_err(|error| error.to_string())? - { + if !is_private_ad_chat(chat_id, user_id) { return Ok(()); } + let delivery = ImageAdDelivery::Persistent; let source_id = job_id.to_string(); let Some(ad) = self .ads .prepare(GradiusUtilityAdRequest { surface: GradiusUtilitySurface::Image, - include_vip_appendix: chat_id > 0 || delivery == ImageAdDelivery::Ephemeral, + include_vip_appendix: true, source_id: source_id.clone(), attempt_key, user_id, @@ -99,29 +91,6 @@ pub enum ImageAdDelivery { Ephemeral, } -async fn image_ad_delivery( - api: &Api, - bot_id: i64, - chat_id: i64, -) -> Result { - if chat_id > 0 { - return Ok(ImageAdDelivery::Persistent); - } - let member = api - .get_chat_member(chat_id, bot_id) - .await - .map_err(|_| "Failed to resolve bot membership for image advertising".to_owned())?; - if member.get_user().id != bot_id || !member.get_user().is_bot || !member.is_member() { - return Err("Image advertising requires active bot membership".to_owned()); - } - Ok(match member { - carapax::types::ChatMember::Administrator(_) | carapax::types::ChatMember::Creator(_) => { - ImageAdDelivery::Ephemeral - } - _ => ImageAdDelivery::Persistent, - }) -} - #[derive(Clone, Debug)] pub struct GradiusUtilityOutbox { store: PostgresGradiusAdStore, @@ -265,6 +234,9 @@ impl GradiusUtilityOutbox { method: TelegramOutboundMethod, job_id: Option, ) -> Result, String> { + if chat_id <= 0 { + return Err("Advertising is restricted to private chats".to_owned()); + } let command = match OutboundCommand::try_from_method(method) { Ok(command) => command, Err(error) => { @@ -360,8 +332,8 @@ pub(crate) fn image_ad_message( reply_to_message_id: i64, html: &str, ) -> Result { - if user_id <= 0 || chat_id == 0 || (chat_id > 0 && chat_id != user_id) { - return Err("Image advertising requires an identified initiator".to_owned()); + if !is_private_ad_chat(chat_id, user_id) { + return Err("Advertising requires the initiator's private chat".to_owned()); } if delivery == ImageAdDelivery::Ephemeral && chat_id > 0 { return Err("Ephemeral image advertising requires a group chat".to_owned()); @@ -423,6 +395,9 @@ fn utility_rich_send( reply_to: i64, html: &str, ) -> Result { + if chat_id <= 0 { + return Err("Advertising is restricted to private chats".to_owned()); + } Ok(SendRichMessage { chat_id, html: utility_rich_html(html)?, @@ -441,6 +416,9 @@ fn utility_rich_edit( message_id: i64, html: &str, ) -> Result { + if chat_id <= 0 { + return Err("Advertising is restricted to private chats".to_owned()); + } Ok(EditRichMessage { chat_id, message_id, @@ -455,11 +433,26 @@ mod tests { use super::*; #[test] - fn image_ads_preserve_the_selected_visibility_on_replay() { + fn all_ad_delivery_methods_reject_non_private_chats() { + for chat_id in [-10042, 0] { + assert!( + image_ad_message(chat_id, 42, ImageAdDelivery::Persistent, None, 77, "Offer") + .is_err() + ); + assert!( + image_ad_message(chat_id, 42, ImageAdDelivery::Ephemeral, None, 77, "Offer") + .is_err() + ); + assert!(utility_rich_send(chat_id, None, 77, "

Offer

").is_err()); + assert!(utility_rich_edit(chat_id, 77, "

Offer

").is_err()); + } + } + + #[test] + fn private_image_ads_preserve_their_payload_on_replay() { for (chat, user, delivery, thread) in [ - (-100, 42, ImageAdDelivery::Ephemeral, Some(12)), - (-100, 42, ImageAdDelivery::Persistent, Some(12)), (42, 42, ImageAdDelivery::Persistent, None), + (42, 42, ImageAdDelivery::Persistent, Some(12)), ] { let method = image_ad_message(chat, user, delivery, thread, 77, "📢 Offer") .expect("image ad"); @@ -506,93 +499,6 @@ mod tests { } } - struct MemberApi { - result: Result, - calls: std::sync::Mutex>, - } - - impl crate::settings::GroupSettingsMemberApi for MemberApi { - type Error = &'static str; - - fn get_chat_member<'a>( - &'a self, - chat_id: i64, - user_id: i64, - ) -> crate::settings::GroupSettingsMemberFuture<'a, carapax::types::ChatMember, Self::Error> - { - Box::pin(async move { - self.calls - .lock() - .expect("member calls") - .push((chat_id, user_id)); - self.result.clone() - }) - } - } - - fn member_api(status: &str, user_id: i64, is_bot: bool) -> MemberApi { - let member = serde_json::from_value(serde_json::json!({ - "status":status,"user":{"id":user_id,"is_bot":is_bot,"first_name":"Bot"}, - "is_anonymous":false,"can_be_edited":false,"can_change_info":false, - "can_delete_messages":false,"can_invite_users":false,"can_manage_chat":true, - "can_manage_video_chats":false,"can_promote_members":false, - "can_restrict_members":false,"can_send_welcome_messages":false - })) - .expect("member fixture"); - MemberApi { - result: Ok(member), - calls: std::sync::Mutex::new(Vec::new()), - } - } - - #[tokio::test] - async fn image_ad_visibility_uses_current_bot_membership_and_private_needs_no_lookup() { - for (status, expected) in [ - ("administrator", ImageAdDelivery::Ephemeral), - ("creator", ImageAdDelivery::Ephemeral), - ("member", ImageAdDelivery::Persistent), - ] { - let api = member_api(status, 7, true); - assert_eq!(image_ad_delivery(&api, 7, -100).await, Ok(expected)); - assert_eq!(*api.calls.lock().expect("member calls"), vec![(-100, 7)]); - } - let api = MemberApi { - result: Err("unavailable"), - calls: std::sync::Mutex::new(Vec::new()), - }; - assert_eq!( - image_ad_delivery(&api, 7, 42).await, - Ok(ImageAdDelivery::Persistent) - ); - assert!(api.calls.lock().expect("member calls").is_empty()); - } - - #[tokio::test] - async fn image_ad_visibility_does_not_publish_on_unknown_or_invalid_bot_membership() { - for api in [ - MemberApi { - result: Err("unavailable"), - calls: std::sync::Mutex::new(Vec::new()), - }, - member_api("administrator", 42, true), - member_api("administrator", 7, false), - member_api("left", 7, true), - ] { - assert!(image_ad_delivery(&api, 7, -100).await.is_err()); - } - let mut api = member_api("administrator", 7, true); - assert_eq!( - image_ad_delivery(&api, 7, -100).await, - Ok(ImageAdDelivery::Ephemeral) - ); - api.result = member_api("member", 7, true).result; - assert_eq!( - image_ad_delivery(&api, 7, -100).await, - Ok(ImageAdDelivery::Persistent) - ); - assert_eq!(api.calls.lock().expect("member calls").len(), 2); - } - #[test] fn rich_utility_send_and_edit_preserve_content_and_ad_separator_on_replay() { for content in [ diff --git a/crates/openplotva-app/src/help.rs b/crates/openplotva-app/src/help.rs index cce620cc..2e654495 100644 --- a/crates/openplotva-app/src/help.rs +++ b/crates/openplotva-app/src/help.rs @@ -615,6 +615,7 @@ fn command_for_bot(command: &ParsedBotCommand<'_>, private_chat: bool, bot_usern #[cfg(test)] mod tests { use super::*; + use openplotva_storage::database as sqlx; use openplotva_telegram::build_text_message_methods; use openplotva_updates::{UpdateConsumerConfig, UpdateStageOutcome}; use serde_json::{Value, json}; diff --git a/crates/openplotva-app/src/http/common.rs b/crates/openplotva-app/src/http/common.rs index 1205f670..746c2a33 100644 --- a/crates/openplotva-app/src/http/common.rs +++ b/crates/openplotva-app/src/http/common.rs @@ -4,6 +4,7 @@ use axum::{ http::{HeaderMap, StatusCode, header}, response::{IntoResponse, Response}, }; +use openplotva_storage::database as sqlx; use std::collections::BTreeMap; pub(crate) fn query_values(raw_query: Option<&str>) -> BTreeMap { diff --git a/crates/openplotva-app/src/http/entities.rs b/crates/openplotva-app/src/http/entities.rs index ef21f6c0..309d2273 100644 --- a/crates/openplotva-app/src/http/entities.rs +++ b/crates/openplotva-app/src/http/entities.rs @@ -17,6 +17,7 @@ use axum::{ response::{IntoResponse, Response}, }; use base64::{Engine as _, engine::general_purpose::STANDARD as BASE64_STANDARD}; +use openplotva_storage::database as sqlx; use openplotva_storage::{ PostgresChatMemberStore, PostgresChatSettingsStore, PostgresPaymentStore, PostgresVipStore, RedisBlockedChatStore, diff --git a/crates/openplotva-app/src/http/mod.rs b/crates/openplotva-app/src/http/mod.rs index a970ec7c..78cc502f 100644 --- a/crates/openplotva-app/src/http/mod.rs +++ b/crates/openplotva-app/src/http/mod.rs @@ -1,5 +1,6 @@ //! HTTP state and feature-specific request adapters. +use openplotva_storage::database as sqlx; pub(crate) mod assets; pub(crate) mod auth; pub(crate) mod common; diff --git a/crates/openplotva-app/src/http/observability.rs b/crates/openplotva-app/src/http/observability.rs index 25e06def..0064f5f6 100644 --- a/crates/openplotva-app/src/http/observability.rs +++ b/crates/openplotva-app/src/http/observability.rs @@ -20,6 +20,7 @@ use axum::{ }, }; use futures_util::{Stream, stream}; +use openplotva_storage::database as sqlx; use serde::{Deserialize, Serialize}; use sqlx::PgPool; use std::{collections::VecDeque, convert::Infallible, sync::Arc, time::Duration}; diff --git a/crates/openplotva-app/src/http/routing.rs b/crates/openplotva-app/src/http/routing.rs index 0459a8fd..4e0fa123 100644 --- a/crates/openplotva-app/src/http/routing.rs +++ b/crates/openplotva-app/src/http/routing.rs @@ -14,6 +14,7 @@ use axum::{ http::{HeaderMap, Method, StatusCode}, response::Response, }; +use openplotva_storage::database as sqlx; use serde_json::json; use sqlx::PgPool; use std::{sync::Arc, time::Duration}; diff --git a/crates/openplotva-app/src/lib.rs b/crates/openplotva-app/src/lib.rs index 27bcb879..86611242 100644 --- a/crates/openplotva-app/src/lib.rs +++ b/crates/openplotva-app/src/lib.rs @@ -1,5 +1,6 @@ //! Composition root for the OpenPlotva application shell. +use openplotva_storage::database as sqlx; pub mod admin; pub mod agent_runtime; mod agent_tools; @@ -4541,8 +4542,6 @@ async fn start_runtime_workers( gradius_utility_outbox::GradiusUtilityImageAds { ads: Arc::clone(ads), outbox: Arc::clone(outbox), - telegram: telegram.clone(), - bot_id: bot_identity.id, }, )); } @@ -5668,6 +5667,7 @@ mod tests { admin_auth_check, admin_auth_date_is_fresh, admin_auth_response, admin_auth_user_state, admin_bootstrap, admin_session_is_authorized, admin_static_asset_requires_auth, }; + use openplotva_storage::database as sqlx; use crate::http::common::parse_memory_limit; use crate::http::settings::{ diff --git a/crates/openplotva-app/src/maintenance_api.rs b/crates/openplotva-app/src/maintenance_api.rs index 506d5ce3..b8372e76 100644 --- a/crates/openplotva-app/src/maintenance_api.rs +++ b/crates/openplotva-app/src/maintenance_api.rs @@ -1,3 +1,4 @@ +use openplotva_storage::database as sqlx; use std::{sync::Arc, time::Duration}; use axum::{ @@ -417,6 +418,7 @@ pub async fn record_dispatch( #[cfg(test)] mod tests { use super::*; + use openplotva_storage::database as sqlx; #[tokio::test] async fn maintenance_lifecycle_captures_without_telegram_and_retains_notifications() diff --git a/crates/openplotva-app/src/model_routing.rs b/crates/openplotva-app/src/model_routing.rs index 37fe4e07..c208a13c 100644 --- a/crates/openplotva-app/src/model_routing.rs +++ b/crates/openplotva-app/src/model_routing.rs @@ -5,6 +5,7 @@ //! unit-tested without a database; [`load_routing_table`] and [`reload_router`] //! wrap it with the storage read and the atomic `ArcSwap` publish. +use openplotva_storage::database as sqlx; use std::collections::HashMap; use std::sync::Arc; use std::time::Duration; @@ -2589,6 +2590,7 @@ fn build_route( #[cfg(test)] mod tests { use openplotva_llm::router::{BreakerLiveness, BreakerSet, TriggerState, select}; + use openplotva_storage::database as sqlx; use openplotva_storage::llm_routing::{ AssignmentRecord, ModelRecord, PoolRecord, ProviderRecord, RoutingSnapshot, TriggerRecord, WorkflowRecord, diff --git a/crates/openplotva-app/src/music_jobs.rs b/crates/openplotva-app/src/music_jobs.rs index 9320d244..8c8efe76 100644 --- a/crates/openplotva-app/src/music_jobs.rs +++ b/crates/openplotva-app/src/music_jobs.rs @@ -1,5 +1,6 @@ //! App-level music generation task execution. +use openplotva_storage::database as sqlx; use std::{ collections::BTreeMap, fmt, future::Future, pin::Pin, sync::Arc, time::Duration as StdDuration, }; diff --git a/crates/openplotva-app/src/openrouter_free_pool.rs b/crates/openplotva-app/src/openrouter_free_pool.rs index 8e16646c..0c9270a0 100644 --- a/crates/openplotva-app/src/openrouter_free_pool.rs +++ b/crates/openplotva-app/src/openrouter_free_pool.rs @@ -1,3 +1,4 @@ +use openplotva_storage::database as sqlx; use std::{ collections::{HashMap, HashSet}, sync::Mutex, diff --git a/crates/openplotva-app/src/rates.rs b/crates/openplotva-app/src/rates.rs index 3c3214ed..b49d8e70 100644 --- a/crates/openplotva-app/src/rates.rs +++ b/crates/openplotva-app/src/rates.rs @@ -2019,6 +2019,7 @@ fn user_full_name(user: &TelegramUser) -> String { mod tests { use super::*; use crate::updates::UpdateHandler; + use openplotva_storage::database as sqlx; use serde_json::json; use std::{ io::{self, Read, Write}, diff --git a/crates/openplotva-app/src/rich.rs b/crates/openplotva-app/src/rich.rs index adf95907..34b54015 100644 --- a/crates/openplotva-app/src/rich.rs +++ b/crates/openplotva-app/src/rich.rs @@ -1,5 +1,7 @@ //! Rich layouts, durable final delivery, and temporary draft transport. +#[cfg(test)] +use openplotva_storage::database as sqlx; use openplotva_telegram::{escape_telegram_html_text as esc, format_rich_html, sanitize_rich_html}; /// Persists final rich messages through the existing Telegram outbox. diff --git a/crates/openplotva-app/src/runtime_analytics_overview.rs b/crates/openplotva-app/src/runtime_analytics_overview.rs index 88ce32ff..ca223e8c 100644 --- a/crates/openplotva-app/src/runtime_analytics_overview.rs +++ b/crates/openplotva-app/src/runtime_analytics_overview.rs @@ -8,6 +8,7 @@ //! honest (no rollup approximation). Long-range historical trends stay on the //! rollup-backed views elsewhere. +use openplotva_storage::database as sqlx; use std::time::Duration as StdDuration; use serde_json::{Value, json}; diff --git a/crates/openplotva-app/src/runtime_entities.rs b/crates/openplotva-app/src/runtime_entities.rs index 157c26a5..1d5cbbb9 100644 --- a/crates/openplotva-app/src/runtime_entities.rs +++ b/crates/openplotva-app/src/runtime_entities.rs @@ -7,6 +7,7 @@ use openplotva_server::{ RuntimeUserDetailsData, RuntimeUserLookup, RuntimeUsersFilter, RuntimeVipCacheData, RuntimeVipEventData, RuntimeVipSummaryData, }; +use openplotva_storage::database as sqlx; use openplotva_storage::entities::{ ChatMemberProfile, PostgresEntityStore, TelegramChatProfile, TelegramUserProfile, }; @@ -459,6 +460,7 @@ fn error_text(error: openplotva_storage::StorageError) -> String { #[cfg(test)] mod tests { use super::*; + use openplotva_storage::database as sqlx; #[test] fn subscription_status_matches_go_priority_order() { diff --git a/crates/openplotva-app/src/runtime_gpu_arbiter.rs b/crates/openplotva-app/src/runtime_gpu_arbiter.rs index e8ba3a14..e5b72093 100644 --- a/crates/openplotva-app/src/runtime_gpu_arbiter.rs +++ b/crates/openplotva-app/src/runtime_gpu_arbiter.rs @@ -4,6 +4,7 @@ //! the panel degrades per-section instead of blanking when the butler or the //! database is unreachable. +use openplotva_storage::database as sqlx; use std::collections::BTreeMap; use std::time::Duration as StdDuration; diff --git a/crates/openplotva-app/src/runtime_gradius.rs b/crates/openplotva-app/src/runtime_gradius.rs index 183a8fe3..b2f32d2f 100644 --- a/crates/openplotva-app/src/runtime_gradius.rs +++ b/crates/openplotva-app/src/runtime_gradius.rs @@ -4,6 +4,7 @@ use openplotva_server::{ RuntimeGradiusAuditReader, RuntimeGradiusAuditReaderFuture, RuntimeGradiusOpportunitiesFilter, RuntimeGradiusSummaryFilter, }; +use openplotva_storage::database as sqlx; use serde_json::{Value, json}; use sqlx::{PgPool, Postgres, QueryBuilder, Row}; use time::{Duration, OffsetDateTime, format_description::well_known::Rfc3339}; @@ -311,6 +312,7 @@ fn error_text(error: impl std::fmt::Display) -> String { #[cfg(test)] mod tests { + use openplotva_storage::database as sqlx; use std::{env, error::Error}; use openplotva_storage::gradius_ads::{ diff --git a/crates/openplotva-app/src/runtime_llm.rs b/crates/openplotva-app/src/runtime_llm.rs index e13d0963..da1724cb 100644 --- a/crates/openplotva-app/src/runtime_llm.rs +++ b/crates/openplotva-app/src/runtime_llm.rs @@ -1,5 +1,6 @@ //! Runtime API LLM trace buffer and dialog provider recorder. +use openplotva_storage::database as sqlx; use std::{ sync::{ Arc, Mutex, @@ -1719,6 +1720,7 @@ fn format_ts(ts: OffsetDateTime) -> String { #[cfg(test)] mod tests { use openplotva_server::RuntimeLlmGenConfigData; + use openplotva_storage::database as sqlx; use super::*; diff --git a/crates/openplotva-app/src/runtime_llm_analytics.rs b/crates/openplotva-app/src/runtime_llm_analytics.rs index 2bf1409a..c868ef7b 100644 --- a/crates/openplotva-app/src/runtime_llm_analytics.rs +++ b/crates/openplotva-app/src/runtime_llm_analytics.rs @@ -1,5 +1,6 @@ //! Runtime API LLM analytics reader backed by Postgres. +use openplotva_storage::database as sqlx; use std::{collections::BTreeMap, sync::Arc, time::Duration as StdDuration}; use openplotva_server::{ diff --git a/crates/openplotva-app/src/runtime_retention.rs b/crates/openplotva-app/src/runtime_retention.rs index d4aacde1..214c1c65 100644 --- a/crates/openplotva-app/src/runtime_retention.rs +++ b/crates/openplotva-app/src/runtime_retention.rs @@ -9,6 +9,7 @@ //! sleep. The drain inner loop clears a large initial backlog gradually (a small //! pause between batches) so it does not spike replication lag or autovacuum. +use openplotva_storage::database as sqlx; use std::time::Duration; use openplotva_storage::PostgresTelegramDeliveryStore; diff --git a/crates/openplotva-app/src/runtime_routing.rs b/crates/openplotva-app/src/runtime_routing.rs index 7408b6ec..d83cd61c 100644 --- a/crates/openplotva-app/src/runtime_routing.rs +++ b/crates/openplotva-app/src/runtime_routing.rs @@ -1,3 +1,4 @@ +use openplotva_storage::database as sqlx; use std::{ sync::{ Arc, Mutex, diff --git a/crates/openplotva-app/src/runtime_safety.rs b/crates/openplotva-app/src/runtime_safety.rs index 5646b31f..892e695e 100644 --- a/crates/openplotva-app/src/runtime_safety.rs +++ b/crates/openplotva-app/src/runtime_safety.rs @@ -1,5 +1,6 @@ //! Runtime API WhiteCircle safety-check reader backed by Postgres. +use openplotva_storage::database as sqlx; use std::time::Duration; use openplotva_server::{ diff --git a/crates/openplotva-app/src/runtime_sql.rs b/crates/openplotva-app/src/runtime_sql.rs index 21084712..6143adcc 100644 --- a/crates/openplotva-app/src/runtime_sql.rs +++ b/crates/openplotva-app/src/runtime_sql.rs @@ -1,3 +1,4 @@ +use openplotva_storage::database as sqlx; use std::time::{Duration, Instant}; use futures_util::TryStreamExt as _; diff --git a/crates/openplotva-app/src/runtime_telegram_delivery.rs b/crates/openplotva-app/src/runtime_telegram_delivery.rs index 6a86aa1b..887a637a 100644 --- a/crates/openplotva-app/src/runtime_telegram_delivery.rs +++ b/crates/openplotva-app/src/runtime_telegram_delivery.rs @@ -8,6 +8,7 @@ use openplotva_server::{ RuntimeTelegramUpdateAttemptData, RuntimeTelegramUpdateInboxItemData, RuntimeTelegramUpdateInboxStatsData, }; +use openplotva_storage::database as sqlx; use openplotva_storage::{ PostgresTelegramDeliveryStore, PostgresTelegramOutboxStore, TelegramOutboxAttempt, TelegramOutboxItem, TelegramOutboxStats, TelegramUpdateAttempt, TelegramUpdateInboxItem, diff --git a/crates/openplotva-app/src/runtime_updates.rs b/crates/openplotva-app/src/runtime_updates.rs index cf8ddf96..349fbd02 100644 --- a/crates/openplotva-app/src/runtime_updates.rs +++ b/crates/openplotva-app/src/runtime_updates.rs @@ -1,3 +1,4 @@ +use openplotva_storage::database as sqlx; use std::{ collections::{HashMap, VecDeque}, sync::{ diff --git a/crates/openplotva-app/src/telegram_outbox.rs b/crates/openplotva-app/src/telegram_outbox.rs index aeb4fb1b..cf151ee2 100644 --- a/crates/openplotva-app/src/telegram_outbox.rs +++ b/crates/openplotva-app/src/telegram_outbox.rs @@ -5,6 +5,7 @@ //! never replayed automatically: lease recovery moves it to `ambiguous` and an //! idempotent terminal reaction is queued instead. +use openplotva_storage::database as sqlx; /// Identity returned after all outbound parts were accepted by one queue. #[derive(Clone, Debug, Eq, PartialEq)] pub struct QueuedBatchReceipt { @@ -1915,6 +1916,7 @@ fn merge_worker_report( #[cfg(test)] mod tests { use super::*; + use openplotva_storage::database as sqlx; struct UtilityReceiptTransport { already_edited: bool, @@ -2111,6 +2113,18 @@ mod tests { html: "

Winner

📢 Offer".to_owned(), reply_markup: None, }) + } else if generation_group { + let mut payload = json!({"chat_id": chat_id, "text": "📢 Offer", "parse_mode": "HTML", "link_preview_options": {"is_disabled": true}, "reply_parameters": {"message_id": 77, "allow_sending_without_reply": false}}); + if ephemeral { + payload["ephemeral_message_parameters"] = + json!({"receiver_user_id": 9_820_825_960_i64}); + } + OutboundCommand::decode( + openplotva_telegram::OUTBOUND_COMMAND_PAYLOAD_VERSION, + "sendMessage", + &serde_json::to_vec(&payload)?, + )? + .into_method() } else { TelegramOutboundMethod::from(crate::gradius_utility_outbox::image_ad_message( chat_id, diff --git a/crates/openplotva-app/src/update_materializer.rs b/crates/openplotva-app/src/update_materializer.rs index 4f785b0a..afb81084 100644 --- a/crates/openplotva-app/src/update_materializer.rs +++ b/crates/openplotva-app/src/update_materializer.rs @@ -2134,6 +2134,7 @@ where #[cfg(test)] mod tests { + use openplotva_storage::database as sqlx; use std::{ collections::{BTreeMap, VecDeque}, env, diff --git a/crates/openplotva-storage/Cargo.toml b/crates/openplotva-storage/Cargo.toml index 6f0d919d..539c69d7 100644 --- a/crates/openplotva-storage/Cargo.toml +++ b/crates/openplotva-storage/Cargo.toml @@ -21,12 +21,16 @@ redis = { workspace = true, features = ["tokio-comp", "connection-manager"] } serde = { workspace = true, features = ["derive"] } serde_json.workspace = true sha2.workspace = true -sqlx = { workspace = true, features = ["macros", "migrate", "postgres", "runtime-tokio", "time"] } +sqlx-core = { workspace = true, features = ["json", "migrate", "offline", "_rt-tokio", "time"] } +sqlx-postgres = { workspace = true, features = ["json", "migrate", "offline", "time"] } thiserror.workspace = true time.workspace = true tokio = { workspace = true, features = ["sync", "time"] } tracing.workspace = true +[build-dependencies] +sqlx-core = { workspace = true, features = ["migrate"] } + [dev-dependencies] tokio = { workspace = true, features = ["macros", "rt-multi-thread"] } diff --git a/crates/openplotva-storage/build.rs b/crates/openplotva-storage/build.rs new file mode 100644 index 00000000..7a5d934a --- /dev/null +++ b/crates/openplotva-storage/build.rs @@ -0,0 +1,40 @@ +use std::{env, error::Error, fmt::Write, fs, path::PathBuf}; + +use sqlx_core::migrate::{MigrationType, resolve_blocking}; + +fn main() -> Result<(), Box> { + let manifest_dir = PathBuf::from(env::var("CARGO_MANIFEST_DIR")?); + let migrations_dir = manifest_dir.join("../../migrations").canonicalize()?; + println!("cargo:rerun-if-changed={}", migrations_dir.display()); + + let mut output = String::from( + "pub static MIGRATOR: crate::database::migrate::Migrator = crate::database::migrate::Migrator {\n\ + migrations: std::borrow::Cow::Borrowed(&[\n", + ); + for (migration, path) in resolve_blocking(&migrations_dir)? { + let migration_type = match migration.migration_type { + MigrationType::Simple => "Simple", + MigrationType::ReversibleUp => "ReversibleUp", + MigrationType::ReversibleDown => "ReversibleDown", + }; + let path = path.canonicalize()?; + let path = path.to_str().ok_or("Migration path is not UTF-8")?; + writeln!( + output, + "crate::database::migrate::Migration {{ version: {}, description: std::borrow::Cow::Borrowed({:?}), \ + migration_type: crate::database::migrate::MigrationType::{}, \ + sql: crate::database::SqlStr::from_static(include_str!({:?})), \ + checksum: std::borrow::Cow::Borrowed(&{:?}), no_tx: {} }},", + migration.version, + migration.description, + migration_type, + path, + migration.checksum, + migration.no_tx, + )?; + } + output.push_str("]), ..crate::database::migrate::Migrator::DEFAULT };\n"); + let destination = PathBuf::from(env::var("OUT_DIR")?).join("migrations.rs"); + fs::write(destination, output)?; + Ok(()) +} diff --git a/crates/openplotva-storage/src/agent_tools.rs b/crates/openplotva-storage/src/agent_tools.rs index 07ce8f99..4f6562e6 100644 --- a/crates/openplotva-storage/src/agent_tools.rs +++ b/crates/openplotva-storage/src/agent_tools.rs @@ -1,4 +1,5 @@ //! Caller-scoped queries used by dialog tools. +use crate::database as sqlx; use crate::{PostgresHistoryStore, PostgresMemoryStore, StorageError}; use serde_json::{Value, json}; use sqlx::Row; @@ -140,6 +141,7 @@ impl PostgresMemoryStore { #[cfg(test)] mod tests { use super::*; + use crate::database as sqlx; #[tokio::test] async fn live_agent_tools_enforce_owner_topic_reset_and_memory_scope() diff --git a/crates/openplotva-storage/src/app_settings.rs b/crates/openplotva-storage/src/app_settings.rs index fd207c91..e897f053 100644 --- a/crates/openplotva-storage/src/app_settings.rs +++ b/crates/openplotva-storage/src/app_settings.rs @@ -1,6 +1,7 @@ //! Persisted application settings, usable in an existing transaction. use crate::StorageError; +use crate::database as sqlx; use sqlx::{Executor, Postgres}; pub async fn app_setting<'a>( diff --git a/crates/openplotva-storage/src/database.rs b/crates/openplotva-storage/src/database.rs new file mode 100644 index 00000000..7cc1403f --- /dev/null +++ b/crates/openplotva-storage/src/database.rs @@ -0,0 +1,24 @@ +//! PostgreSQL-only SQLx API shared with runtime adapters. + +pub use sqlx_core::{ + acquire::Acquire, + column::Column, + connection::Connection, + database::Database, + error::{self, Error}, + executor::{Execute, Executor}, + from_row::FromRow, + migrate, + query::{self, query}, + query_as::query_as, + query_builder::QueryBuilder, + query_scalar::query_scalar, + raw_sql::raw_sql, + row::Row, + sql_str::{AssertSqlSafe, SqlSafeStr, SqlStr}, + transaction::Transaction, + type_info::TypeInfo, + types, + value::ValueRef, +}; +pub use sqlx_postgres::{self as postgres, PgConnection, PgPool, Postgres}; diff --git a/crates/openplotva-storage/src/entities.rs b/crates/openplotva-storage/src/entities.rs index 47acf8f1..8661e263 100644 --- a/crates/openplotva-storage/src/entities.rs +++ b/crates/openplotva-storage/src/entities.rs @@ -1,10 +1,11 @@ //! Stored Telegram entity profiles shared by HTTP and runtime API adapters. use crate::StorageError; +use crate::database as sqlx; use sqlx::{FromRow, PgPool, Row}; use time::OffsetDateTime; -#[derive(Clone, Debug, serde::Deserialize, sqlx::FromRow)] +#[derive(Clone, Debug, serde::Deserialize)] pub struct TelegramUserProfile { pub id: i64, pub is_premium: Option, @@ -20,10 +21,26 @@ pub struct TelegramUserProfile { pub updated: Option, } -#[derive(Clone, Debug, sqlx::FromRow)] +impl<'r> sqlx::FromRow<'r, sqlx::postgres::PgRow> for TelegramUserProfile { + fn from_row(row: &'r sqlx::postgres::PgRow) -> Result { + Ok(Self { + id: row.try_get("id")?, + is_premium: row.try_get("is_premium")?, + first_name: row.try_get("first_name")?, + last_name: row.try_get("last_name")?, + username: row.try_get("username")?, + language_code: row.try_get("language_code")?, + is_vip: row.try_get("is_vip")?, + settings: row.try_get("settings")?, + discovered: row.try_get("discovered")?, + updated: row.try_get("updated")?, + }) + } +} + +#[derive(Clone, Debug)] pub struct TelegramChatProfile { pub id: i64, - #[sqlx(rename = "type")] pub chat_type: String, pub title: Option, pub username: Option, @@ -55,6 +72,44 @@ pub struct TelegramChatProfile { pub updated: Option, } +impl<'r> sqlx::FromRow<'r, sqlx::postgres::PgRow> for TelegramChatProfile { + fn from_row(row: &'r sqlx::postgres::PgRow) -> Result { + Ok(Self { + id: row.try_get("id")?, + chat_type: row.try_get("type")?, + title: row.try_get("title")?, + username: row.try_get("username")?, + first_name: row.try_get("first_name")?, + last_name: row.try_get("last_name")?, + is_forum: row.try_get("is_forum")?, + active_usernames: row.try_get("active_usernames")?, + available_reactions: row.try_get("available_reactions")?, + bio: row.try_get("bio")?, + has_private_forwards: row.try_get("has_private_forwards")?, + has_restricted_voice_and_video_messages: row + .try_get("has_restricted_voice_and_video_messages")?, + join_to_send_messages: row.try_get("join_to_send_messages")?, + join_by_request: row.try_get("join_by_request")?, + description: row.try_get("description")?, + invite_link: row.try_get("invite_link")?, + pinned_message: row.try_get("pinned_message")?, + permissions: row.try_get("permissions")?, + slow_mode_delay: row.try_get("slow_mode_delay")?, + message_auto_delete_time: row.try_get("message_auto_delete_time")?, + has_aggressive_anti_spam_enabled: row.try_get("has_aggressive_anti_spam_enabled")?, + has_hidden_members: row.try_get("has_hidden_members")?, + has_protected_content: row.try_get("has_protected_content")?, + has_visible_history: row.try_get("has_visible_history")?, + sticker_set_name: row.try_get("sticker_set_name")?, + can_set_sticker_set: row.try_get("can_set_sticker_set")?, + linked_chat_id: row.try_get("linked_chat_id")?, + location: row.try_get("location")?, + discovered: row.try_get("discovered")?, + updated: row.try_get("updated")?, + }) + } +} + const SQL_COUNT_USERS_FILTERED: &str = r#" SELECT COUNT(*)::int FROM telegram_users_effective @@ -236,9 +291,8 @@ const SQL_ADMIN_LIST_VIP_USERS: &str = "WITH latest_vip AS (SELECT DISTINCT ON ( const SQL_ADMIN_SEARCH_CHATS_BY_MEMBER: &str = "SELECT DISTINCT c.* FROM telegram_chats_effective c JOIN telegram_chat_members_effective cm ON c.id = cm.chat_id JOIN telegram_users_effective u ON cm.user_id = u.id WHERE ($1::text IS NULL OR LOWER(u.username) = LOWER($1::text)) AND ($2::bigint IS NULL OR u.id = $2::bigint) ORDER BY c.id LIMIT $3"; -#[derive(Clone, Debug, sqlx::FromRow)] +#[derive(Clone, Debug)] pub struct VipUserProfile { - #[sqlx(flatten)] pub user: TelegramUserProfile, pub remaining_seconds: Option, pub vip_active: Option, @@ -252,6 +306,24 @@ pub struct VipUserProfile { pub latest_subscription_expires_at: Option, } +impl<'r> sqlx::FromRow<'r, sqlx::postgres::PgRow> for VipUserProfile { + fn from_row(row: &'r sqlx::postgres::PgRow) -> Result { + Ok(Self { + user: >::from_row(row)?, + remaining_seconds: row.try_get("remaining_seconds")?, + vip_active: row.try_get("vip_active")?, + vip_expires_at: row.try_get("vip_expires_at")?, + latest_event_id: row.try_get("latest_event_id")?, + latest_event_type: row.try_get("latest_event_type")?, + latest_delta_seconds: row.try_get("latest_delta_seconds")?, + latest_reason: row.try_get("latest_reason")?, + latest_created_at: row.try_get("latest_created_at")?, + subscriptions_count: row.try_get("subscriptions_count")?, + latest_subscription_expires_at: row.try_get("latest_subscription_expires_at")?, + }) + } +} + impl PostgresEntityStore { pub async fn vip_users( &self, @@ -283,15 +355,26 @@ impl PostgresEntityStore { } } -#[derive(Clone, Debug, sqlx::FromRow)] +#[derive(Clone, Debug)] pub struct ChatMemberProfile { - #[sqlx(flatten)] pub membership: crate::ChatMemberRecord, pub created_at: Option, pub updated_at: Option, pub last_message_at: Option, } +impl<'r> sqlx::FromRow<'r, sqlx::postgres::PgRow> for ChatMemberProfile { + fn from_row(row: &'r sqlx::postgres::PgRow) -> Result { + Ok(Self { + membership: + >::from_row(row)?, + created_at: row.try_get("created_at")?, + updated_at: row.try_get("updated_at")?, + last_message_at: row.try_get("last_message_at")?, + }) + } +} + impl PostgresEntityStore { pub async fn members_with_users( &self, @@ -341,7 +424,7 @@ impl PostgresEntityStore { } } -#[derive(Clone, Debug, sqlx::FromRow)] +#[derive(Clone, Debug)] pub struct StoredChatSettings { pub chat_id: i64, pub mood_alignment: Option, @@ -359,7 +442,28 @@ pub struct StoredChatSettings { pub greeting_html: Option, } -#[derive(Clone, Debug, sqlx::FromRow)] +impl<'r> sqlx::FromRow<'r, sqlx::postgres::PgRow> for StoredChatSettings { + fn from_row(row: &'r sqlx::postgres::PgRow) -> Result { + Ok(Self { + chat_id: row.try_get("chat_id")?, + mood_alignment: row.try_get("mood_alignment")?, + custom_persona: row.try_get("custom_persona")?, + updated: row.try_get("updated")?, + reactivity_percentage: row.try_get("reactivity_percentage")?, + proactivity_percentage: row.try_get("proactivity_percentage")?, + enable_global_text_reply: row.try_get("enable_global_text_reply")?, + enable_global_draw_reply: row.try_get("enable_global_draw_reply")?, + enable_obscenifier: row.try_get("enable_obscenifier")?, + enable_profanity: row.try_get("enable_profanity")?, + enable_greet_joiners: row.try_get("enable_greet_joiners")?, + enable_daily_game: row.try_get("enable_daily_game")?, + daily_game_theme: row.try_get("daily_game_theme")?, + greeting_html: row.try_get("greeting_html")?, + }) + } +} + +#[derive(Clone, Debug)] pub struct StoredChatPermissions { pub chat_id: i64, pub status: Option, @@ -387,6 +491,37 @@ pub struct StoredChatPermissions { pub error_count: Option, } +impl<'r> sqlx::FromRow<'r, sqlx::postgres::PgRow> for StoredChatPermissions { + fn from_row(row: &'r sqlx::postgres::PgRow) -> Result { + Ok(Self { + chat_id: row.try_get("chat_id")?, + status: row.try_get("status")?, + can_manage_chat: row.try_get("can_manage_chat")?, + can_delete_messages: row.try_get("can_delete_messages")?, + can_manage_video_chats: row.try_get("can_manage_video_chats")?, + can_restrict_members: row.try_get("can_restrict_members")?, + can_promote_members: row.try_get("can_promote_members")?, + can_change_info: row.try_get("can_change_info")?, + can_invite_users: row.try_get("can_invite_users")?, + can_post_messages: row.try_get("can_post_messages")?, + can_edit_messages: row.try_get("can_edit_messages")?, + can_pin_messages: row.try_get("can_pin_messages")?, + can_manage_topics: row.try_get("can_manage_topics")?, + can_send_messages: row.try_get("can_send_messages")?, + can_send_media_messages: row.try_get("can_send_media_messages")?, + can_send_polls: row.try_get("can_send_polls")?, + can_send_other_messages: row.try_get("can_send_other_messages")?, + can_add_web_page_previews: row.try_get("can_add_web_page_previews")?, + last_checked_at: row.try_get("last_checked_at")?, + last_error_at: row.try_get("last_error_at")?, + error_message: row.try_get("error_message")?, + created_at: row.try_get("created_at")?, + updated_at: row.try_get("updated_at")?, + error_count: row.try_get("error_count")?, + }) + } +} + const SQL_ADMIN_GET_CHAT_SETTINGS: &str = "SELECT * FROM chat_settings WHERE chat_id = $1"; const SQL_ADMIN_GET_CHAT_PERMISSIONS: &str = @@ -421,3 +556,94 @@ impl PostgresEntityStore { .await?) } } + +#[cfg(test)] +mod tests { + use super::*; + use crate::database as sqlx; + + #[tokio::test] + async fn postgres_entity_rows_preserve_field_names_flattening_and_defaults() + -> 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?; + crate::run_migrations_on(&pool).await?; + let user_json = serde_json::json!({"id": 42, "first_name": "Row test", "is_vip": false, "settings": {"language": "ru"}}); + let user: TelegramUserProfile = sqlx::query_as( + "SELECT * FROM jsonb_populate_record(NULL::telegram_users_effective, $1)", + ) + .bind(&user_json) + .fetch_one(&pool) + .await?; + assert_eq!(user.id, 42); + assert_eq!(user.first_name, "Row test"); + assert_eq!(user.is_vip, Some(false)); + assert_eq!(user.settings, Some(serde_json::json!({"language": "ru"}))); + assert_eq!(user.updated, None); + + let chat: TelegramChatProfile = sqlx::query_as( + "SELECT * FROM jsonb_populate_record(NULL::telegram_chats_effective, $1)", + ).bind(serde_json::json!({"id": -10042, "type": "supergroup", "title": "Row test", "is_forum": true})).fetch_one(&pool).await?; + assert_eq!(chat.chat_type, "supergroup"); + assert_eq!(chat.id, -10042); + assert_eq!(chat.is_forum, Some(true)); + + let vip: VipUserProfile = sqlx::query_as( + "SELECT profile.*, 60::bigint AS remaining_seconds, true AS vip_active, NULL::timestamptz AS vip_expires_at, 7::bigint AS latest_event_id, 'purchase'::text AS latest_event_type, 60::bigint AS latest_delta_seconds, NULL::text AS latest_reason, NULL::timestamptz AS latest_created_at, 1::bigint AS subscriptions_count, NULL::timestamptz AS latest_subscription_expires_at FROM jsonb_populate_record(NULL::telegram_users_effective, $1) AS profile", + ).bind(&user_json).fetch_one(&pool).await?; + assert_eq!(vip.user.id, 42); + assert_eq!(vip.user.first_name, user.first_name); + assert_eq!(vip.remaining_seconds, Some(60)); + assert_eq!(vip.latest_event_id, Some(7)); + + let membership = serde_json::json!({"chat_id": -10042, "user_id": 42, "status": "member", "can_delete_messages": false}); + let member: ChatMemberProfile = sqlx::query_as( + "SELECT * FROM jsonb_populate_record(NULL::telegram_chat_members_effective, $1)", + ) + .bind(&membership) + .fetch_one(&pool) + .await?; + assert_eq!(member.membership.user_id, 42); + assert_eq!(member.membership.status, "member"); + assert_eq!(member.membership.can_delete_messages, Some(false)); + assert_eq!(member.membership.telegram_observed_at, None); + + let settings: StoredChatSettings = sqlx::query_as( + "SELECT * FROM jsonb_populate_record(NULL::chat_settings, $1)", + ).bind(serde_json::json!({"chat_id": -10042, "reactivity_percentage": 75, "proactivity_percentage": 20, "enable_global_text_reply": true, "enable_global_draw_reply": false, "enable_obscenifier": false, "enable_profanity": true, "enable_greet_joiners": false})).fetch_one(&pool).await?; + assert_eq!(settings.reactivity_percentage, 75); + assert!(settings.enable_global_text_reply); + assert_eq!(settings.daily_game_theme, None); + + let permissions: StoredChatPermissions = sqlx::query_as( + "SELECT * FROM jsonb_populate_record(NULL::chat_permissions, $1)", + ) + .bind(serde_json::json!({"chat_id": -10042, "can_delete_messages": true, "error_count": 0})) + .fetch_one(&pool) + .await?; + assert_eq!(permissions.can_delete_messages, Some(true)); + assert_eq!(permissions.error_count, Some(0)); + + let missing = sqlx::query_as::<_, TelegramUserProfile>("SELECT 42::bigint AS other_id") + .fetch_one(&pool) + .await; + assert!(matches!(missing, Err(sqlx::Error::ColumnNotFound(_)))); + let invalid = sqlx::query_as::<_, TelegramUserProfile>("SELECT 'invalid'::text AS id") + .fetch_one(&pool) + .await; + assert!(matches!(invalid, Err(sqlx::Error::ColumnDecode { .. }))); + let invalid_default = sqlx::query_as::<_, crate::ChatMemberRecord>( + "SELECT member.*, 'invalid'::text AS telegram_observed_at FROM jsonb_populate_record(NULL::telegram_chat_members_effective, $1) AS member", + ).bind(&membership).fetch_one(&pool).await; + assert!(matches!( + invalid_default, + Err(sqlx::Error::ColumnDecode { .. }) + )); + Ok(()) + } +} diff --git a/crates/openplotva-storage/src/gradius_ads.rs b/crates/openplotva-storage/src/gradius_ads.rs index 58a3aa0d..3a470457 100644 --- a/crates/openplotva-storage/src/gradius_ads.rs +++ b/crates/openplotva-storage/src/gradius_ads.rs @@ -1,3 +1,4 @@ +use crate::database as sqlx; use serde_json::Value; use sqlx::{PgPool, Row}; use thiserror::Error; @@ -1362,6 +1363,7 @@ fn parse_ineligibility_reason(value: &str) -> Option { #[cfg(test)] mod tests { + use crate::database as sqlx; use std::{env, error::Error}; use serde_json::{Value, json}; diff --git a/crates/openplotva-storage/src/lib.rs b/crates/openplotva-storage/src/lib.rs index ac46bd61..8f844d01 100644 --- a/crates/openplotva-storage/src/lib.rs +++ b/crates/openplotva-storage/src/lib.rs @@ -1,7 +1,9 @@ //! Storage boundary for Postgres, pgvector, SQLx, Redis, and Dragonfly. +use crate::database as sqlx; mod agent_tools; pub mod app_settings; +pub mod database; pub mod entities; pub mod redis_inspector; pub mod songs; @@ -37,7 +39,7 @@ use serde::{Deserialize, Serialize}; use sha2::{Digest, Sha256}; use sqlx::{ PgPool, Postgres, Row, Transaction, - migrate::{MigrateError, Migration, Migrator}, + migrate::{MigrateError, Migration}, postgres::{PgPoolOptions, PgRow}, }; use thiserror::Error; @@ -1586,7 +1588,7 @@ pub const CHAT_MEMBER_STATUS_LEFT: &str = "left"; pub const CHAT_MEMBER_STATUS_KICKED: &str = "kicked"; -pub static MIGRATOR: Migrator = sqlx::migrate!("../../migrations"); +include!(concat!(env!("OUT_DIR"), "/migrations.rs")); // SQLx's database advisory lock serializes normal transactional migrations, but // concurrent-index migrations must release the transaction. Serialize callers // inside one process as well, otherwise parallel integration tests can deadlock @@ -3601,7 +3603,7 @@ pub struct PostgresChatMemberStore { pool: PgPool, } -#[derive(Clone, Debug, Default, Eq, PartialEq, sqlx::FromRow)] +#[derive(Clone, Debug, Default, Eq, PartialEq)] pub struct ChatMemberRecord { /// Telegram chat ID. pub chat_id: i64, @@ -3651,10 +3653,46 @@ pub struct ChatMemberRecord { /// Optional restricted/kicked expiration. pub until_date: Option, /// Time at which Telegram last supplied this membership state; absent on effective snapshots. - #[sqlx(default)] pub telegram_observed_at: Option, } +impl<'r> sqlx::FromRow<'r, sqlx::postgres::PgRow> for ChatMemberRecord { + fn from_row(row: &'r sqlx::postgres::PgRow) -> Result { + Ok(Self { + chat_id: row.try_get("chat_id")?, + user_id: row.try_get("user_id")?, + status: row.try_get("status")?, + is_member: row.try_get("is_member")?, + is_anonymous: row.try_get("is_anonymous")?, + custom_title: row.try_get("custom_title")?, + can_be_edited: row.try_get("can_be_edited")?, + can_manage_chat: row.try_get("can_manage_chat")?, + can_delete_messages: row.try_get("can_delete_messages")?, + can_manage_video_chats: row.try_get("can_manage_video_chats")?, + can_restrict_members: row.try_get("can_restrict_members")?, + can_promote_members: row.try_get("can_promote_members")?, + can_change_info: row.try_get("can_change_info")?, + can_invite_users: row.try_get("can_invite_users")?, + can_post_messages: row.try_get("can_post_messages")?, + can_edit_messages: row.try_get("can_edit_messages")?, + can_pin_messages: row.try_get("can_pin_messages")?, + can_manage_topics: row.try_get("can_manage_topics")?, + can_send_messages: row.try_get("can_send_messages")?, + can_send_media_messages: row.try_get("can_send_media_messages")?, + can_send_polls: row.try_get("can_send_polls")?, + can_send_other_messages: row.try_get("can_send_other_messages")?, + can_add_web_page_previews: row.try_get("can_add_web_page_previews")?, + until_date: row.try_get("until_date")?, + telegram_observed_at: row.try_get("telegram_observed_at").or_else( + |error| match error { + sqlx::Error::ColumnNotFound(_) => Ok(None), + error => Err(error), + }, + )?, + }) + } +} + #[derive(Clone, Debug, Default, Eq, PartialEq)] pub struct ChatMemberCandidate { /// Telegram user ID. @@ -9561,6 +9599,7 @@ fn vip_event_list_from_row(row: PgRow) -> Result = super::MIGRATOR.iter().collect(); + assert_eq!(embedded.len(), resolved.len()); + for (embedded, (source, _)) in embedded.iter().zip(resolved) { + assert_eq!(embedded.version, source.version); + assert_eq!(embedded.description, source.description); + assert_eq!(embedded.migration_type, source.migration_type); + assert_eq!(embedded.sql.as_str(), source.sql.as_str()); + assert_eq!(embedded.checksum, source.checksum); + assert_eq!(embedded.no_tx, source.no_tx); + } + } + #[test] fn concurrent_index_migrations_are_single_statement_no_tx_files() { for migration in super::MIGRATOR diff --git a/crates/openplotva-storage/src/llm_routing.rs b/crates/openplotva-storage/src/llm_routing.rs index 3c8da032..a3219aee 100644 --- a/crates/openplotva-storage/src/llm_routing.rs +++ b/crates/openplotva-storage/src/llm_routing.rs @@ -6,6 +6,7 @@ //! crate convention of binding `$N::jsonb` from a JSON string and selecting //! `column::text`; `TEXT[]` maps directly to `Vec`. +use crate::database as sqlx; use aes_gcm::aead::{Aead, AeadCore, KeyInit, OsRng}; use aes_gcm::{Aes256Gcm, Nonce}; use serde::Deserialize; @@ -1941,6 +1942,7 @@ pub fn open_key(master_secret: &str, sealed: &[u8]) -> Result Value { #[cfg(test)] mod tests { + use crate::database as sqlx; use std::error::Error; use serde_json::json; diff --git a/crates/openplotva-storage/src/maintenance_notifications.rs b/crates/openplotva-storage/src/maintenance_notifications.rs index ac415bb3..f46a5d5f 100644 --- a/crates/openplotva-storage/src/maintenance_notifications.rs +++ b/crates/openplotva-storage/src/maintenance_notifications.rs @@ -1,3 +1,4 @@ +use crate::database as sqlx; use serde::Serialize; use sqlx::{PgPool, Row}; use time::OffsetDateTime; @@ -133,6 +134,7 @@ fn notification_from_row( #[cfg(test)] mod tests { use super::*; + use crate::database as sqlx; #[tokio::test] async fn live_maintenance_notification_receipts_fence_replay_and_ambiguous_delivery() diff --git a/crates/openplotva-storage/src/songs.rs b/crates/openplotva-storage/src/songs.rs index 3f3b6531..2cdbc1ce 100644 --- a/crates/openplotva-storage/src/songs.rs +++ b/crates/openplotva-storage/src/songs.rs @@ -1,6 +1,7 @@ //! Generated song material, prepared payloads, and confirmed delivery links. use crate::StorageError; +use crate::database as sqlx; use sqlx::{PgPool, Row, postgres::PgRow}; use time::OffsetDateTime; @@ -201,6 +202,7 @@ fn generated_song_from_row(row: PgRow) -> Result String { #[cfg(test)] mod tests { + use crate::database as sqlx; use std::{env, error::Error, time::SystemTime}; use openplotva_core::{ChatState, UserState}; diff --git a/crates/openplotva-storage/src/telegram_outbox.rs b/crates/openplotva-storage/src/telegram_outbox.rs index 0ee0d039..cf4e5c66 100644 --- a/crates/openplotva-storage/src/telegram_outbox.rs +++ b/crates/openplotva-storage/src/telegram_outbox.rs @@ -1,5 +1,6 @@ //! Durable Postgres outbox for Telegram operations. +use crate::database as sqlx; use std::{collections::HashMap, time::Duration}; use serde_json::Value; @@ -1649,6 +1650,7 @@ fn outbox_attempt_from_row(row: PgRow) -> Result