From a6dd96759a925b41fc7af1182c5f5638f377415e Mon Sep 17 00:00:00 2001 From: Beast Date: Mon, 6 Jul 2026 12:43:57 +0800 Subject: [PATCH 1/3] chore: turn off tweet aggregation --- src/main.rs | 4 ---- 1 file changed, 4 deletions(-) diff --git a/src/main.rs b/src/main.rs index 69758dc..00fed09 100644 --- a/src/main.rs +++ b/src/main.rs @@ -102,10 +102,6 @@ async fn main() -> AppResult<()> { error!("HTTP server exited: {:?}", result); result??; } - result = tweet_synchronizer.spawn_tweet_synchronizer() => { - error!("Tweet synchronizer exited: {:?}", result); - result??; - } } Ok(()) From 864792f9456b8d2bc7d5a868ff07177e721e65ac Mon Sep 17 00:00:00 2001 From: Beast Date: Tue, 7 Jul 2026 13:02:26 +0800 Subject: [PATCH 2/3] feat: clean up tweet aggregator related codes --- config/default.toml | 18 - config/example.toml | 18 - config/test.toml | 18 - src/config.rs | 31 - src/db_persistence.rs | 4 - src/errors.rs | 8 - src/main.rs | 21 +- src/metrics.rs | 118 ---- src/models/mod.rs | 1 - src/models/relevant_tweet.rs | 22 - src/models/tweet_pull_usage.rs | 10 - src/repositories/mod.rs | 1 - src/repositories/relevant_tweet.rs | 47 +- src/repositories/tweet_author.rs | 8 - src/repositories/tweet_pull_usage.rs | 265 --------- src/services/alert_service.rs | 91 --- src/services/mod.rs | 3 - src/services/telegram_service.rs | 91 --- src/services/tweet_synchronizer_service.rs | 652 --------------------- src/utils/mod.rs | 1 - src/utils/test_db.rs | 2 +- src/utils/x_url.rs | 3 - 22 files changed, 5 insertions(+), 1428 deletions(-) delete mode 100644 src/models/tweet_pull_usage.rs delete mode 100644 src/repositories/tweet_pull_usage.rs delete mode 100644 src/services/alert_service.rs delete mode 100644 src/services/telegram_service.rs delete mode 100644 src/services/tweet_synchronizer_service.rs delete mode 100644 src/utils/x_url.rs diff --git a/config/default.toml b/config/default.toml index 44cf630..697f989 100644 --- a/config/default.toml +++ b/config/default.toml @@ -27,24 +27,6 @@ callback_url = "http://localhost:3000/api/auth/x/callback" client_id = "WlVrcm4xSEpXQ2l3TURFM3lLZnE6MTpjaQ" client_secret = "lfXc45dZLqYTzP62Ms32EhXinGQzxcIP9TvjJml2B-h0T1nIJK" -[tweet_sync] -api_key = "some-key" -interval_in_hours = 24 -keywords = "quantum" -monthly_limit = 15000 -alert_threshold = 13500 -reset_day = 22 - -[tg_bot] -base_url = "https://api.telegram.org" -chat_id = "-1" -# This is optional, if you don't need message thread just remove it -message_thread_id = "1" -token = "token" - -[alert] -webhook_url = "https://www.webhook_url.com" - [remote_configs] wallet_configs_file = "../wallet_configs/default_configs.json" diff --git a/config/example.toml b/config/example.toml index b6e9fc4..4b8b979 100644 --- a/config/example.toml +++ b/config/example.toml @@ -29,24 +29,6 @@ callback_url = "http://localhost:12345/example/callback" client_id = "example-id" client_secret = "example-secret" -[tweet_sync] -api_key = "some-key" -interval_in_hours = 24 -keywords = "example" -monthly_limit = 15000 -alert_threshold = 13500 -reset_day = 22 - -[tg_bot] -base_url = "https://api.telegram.org" -chat_id = '-1' -# This is optional, if you don't need message thread just remove it -message_thread_id = '1' -token = "token" - -[alert] -webhook_url = "https://www.webhook_url.com" - [remote_configs] wallet_configs_file = "../wallet_configs/default_configs.json" diff --git a/config/test.toml b/config/test.toml index 655bdb8..4006de9 100644 --- a/config/test.toml +++ b/config/test.toml @@ -27,24 +27,6 @@ callback_url = "http://localhost:12345/api/auth/x/callback" client_id = "test-id" client_secret = "test-secret" -[tweet_sync] -api_key = "some-key" -interval_in_hours = 24 -keywords = "test" -monthly_limit = 15000 -alert_threshold = 13500 -reset_day = 22 - -[tg_bot] -base_url = "https://api.telegram.org" -chat_id = '-1' -# This is optional, if you don't need message thread just remove it -message_thread_id = '1' -token = "token" - -[alert] -webhook_url = "https://www.webhook_url.com" - [remote_configs] wallet_configs_file = "../wallet_configs/test_configs.json" diff --git a/src/config.rs b/src/config.rs index 54c1367..3e1c2bd 100644 --- a/src/config.rs +++ b/src/config.rs @@ -3,7 +3,6 @@ use std::path::Path; use axum::http::HeaderValue; use rusx::config::OauthConfig; use serde::{Deserialize, Serialize}; -use tokio::time; #[derive(Debug, Clone, Serialize, Deserialize)] pub struct Config { @@ -13,9 +12,6 @@ pub struct Config { pub logging: LoggingConfig, pub jwt: JwtConfig, pub x_oauth: OauthConfig, - pub tweet_sync: TweetSyncConfig, - pub tg_bot: TelegramBotConfig, - pub alert: AlertConfig, pub remote_configs: RemoteConfigsConfig, pub risk_checker: RiskCheckerConfig, pub exchange_rate: ExchangeRateConfig, @@ -55,29 +51,6 @@ pub struct JwtConfig { pub exp_in_hours: i64, } -#[derive(Debug, Clone, Serialize, Deserialize)] -pub struct TweetSyncConfig { - pub interval_in_hours: u64, - pub keywords: String, - pub api_key: String, - pub monthly_limit: u32, - pub alert_threshold: u32, - pub reset_day: u32, -} - -#[derive(Debug, Clone, Serialize, Deserialize)] -pub struct TelegramBotConfig { - pub base_url: String, - pub token: String, - pub chat_id: String, - pub message_thread_id: Option, -} - -#[derive(Debug, Clone, Serialize, Deserialize)] -pub struct AlertConfig { - pub webhook_url: String, -} - #[derive(Debug, Clone, Serialize, Deserialize)] pub struct RiskCheckerConfig { pub etherscan_api_key: String, @@ -133,10 +106,6 @@ impl Config { chrono::Duration::hours(self.jwt.exp_in_hours) } - pub fn get_tweet_sync_interval(&self) -> time::Duration { - time::Duration::from_secs(self.tweet_sync.interval_in_hours * 3600) - } - pub fn get_cors_allowed_origins(&self) -> Vec { self.server .cors_allowed_origins diff --git a/src/db_persistence.rs b/src/db_persistence.rs index a265cc1..5a652bb 100644 --- a/src/db_persistence.rs +++ b/src/db_persistence.rs @@ -4,7 +4,6 @@ use crate::repositories::admin::AdminRepository; use crate::repositories::raid_quest::RaidQuestRepository; use crate::repositories::relevant_tweet::RelevantTweetRepository; use crate::repositories::tweet_author::TweetAuthorRepository; -use crate::repositories::tweet_pull_usage::TweetPullUsageRepository; use crate::repositories::DbResult; use crate::repositories::{address::AddressRepository, referral::ReferralRepository}; @@ -30,7 +29,6 @@ pub struct DbPersistence { pub relevant_tweets: RelevantTweetRepository, pub tweet_authors: TweetAuthorRepository, pub raid_quests: RaidQuestRepository, - pub tweet_pull_usage: TweetPullUsageRepository, /// Used by the `create_admin` binary and integration tests (not the main server binary). #[allow(dead_code)] @@ -49,7 +47,6 @@ impl DbPersistence { let relevant_tweets = RelevantTweetRepository::new(&pool); let tweet_authors = TweetAuthorRepository::new(&pool); let raid_quests = RaidQuestRepository::new(&pool); - let tweet_pull_usage = TweetPullUsageRepository::new(pool.clone()); Ok(Self { pool, @@ -59,7 +56,6 @@ impl DbPersistence { relevant_tweets, tweet_authors, raid_quests, - tweet_pull_usage, }) } } diff --git a/src/errors.rs b/src/errors.rs index caeac42..c486e4a 100644 --- a/src/errors.rs +++ b/src/errors.rs @@ -39,8 +39,6 @@ pub enum AppError { Http(#[from] axum::http::Error), #[error("Rusx error: {0}")] Rusx(#[from] SdkError), - #[error("Telegram API error: {1}")] - Telegram(u16, String), #[error("Risk checker error: {0}")] RiskChecker(#[from] RiskCheckerError), #[error("Exchange rate error: {0}")] @@ -52,12 +50,6 @@ pub type AppResult = Result; impl IntoResponse for AppError { fn into_response(self) -> Response { let (status, message) = match self { - // --- Telegram --- - AppError::Telegram(code, err) => ( - StatusCode::from_u16(code).unwrap_or(StatusCode::INTERNAL_SERVER_ERROR), - err, - ), - // --- Wallet Feature Flags --- AppError::WalletConfigs(err) => map_wallet_configs_error(err), diff --git a/src/main.rs b/src/main.rs index 00fed09..f41df84 100644 --- a/src/main.rs +++ b/src/main.rs @@ -2,10 +2,7 @@ use crate::{ args::Args, db_persistence::DbPersistence, errors::{AppError, AppResult}, - services::{ - alert_service::AlertService, graphql_client::GraphqlClient, telegram_service::TelegramService, - tweet_synchronizer_service::TweetSynchronizerService, - }, + services::graphql_client::GraphqlClient, }; use clap::Parser; @@ -68,12 +65,7 @@ async fn main() -> AppResult<()> { let server_address = config.get_server_address(); info!("Starting HTTP server on {}", server_address); - let twitter_gateway = Arc::new(RusxGateway::new( - config.x_oauth.clone(), - Some(config.tweet_sync.api_key.clone()), - )?); - let telegram_service = Arc::new(TelegramService::new(config.tg_bot.clone())); - let alert_service = Arc::new(AlertService::new(config.clone(), db.tweet_pull_usage.clone())); + let twitter_gateway = Arc::new(RusxGateway::new(config.x_oauth.clone(), None)?); let server_db = db.clone(); let server_addr_clone = server_address.clone(); let server_config = Arc::new(config.clone()); @@ -87,15 +79,6 @@ async fn main() -> AppResult<()> { info!("🎯 TaskMaster is now running!"); info!("HTTP API available at: http://{}", server_address); - // Initialize tweet sync service - let tweet_synchronizer = TweetSynchronizerService::new( - db.clone(), - twitter_gateway.clone(), - telegram_service, - alert_service.clone(), - Arc::new(config.clone()), - ); - // Wait for any task to complete (they should run forever unless there's an error) tokio::select! { result = server_task => { diff --git a/src/metrics.rs b/src/metrics.rs index 6a933b5..4cea265 100644 --- a/src/metrics.rs +++ b/src/metrics.rs @@ -12,7 +12,6 @@ use std::sync::Arc; use std::time::Instant; use crate::http_server::AppState; -use rusx::error::SdkError; // Define comprehensive metrics for REST API monitoring lazy_static! { @@ -50,35 +49,6 @@ lazy_static! { &["method", "endpoint", "status"] ) .unwrap(); - - // Twitter API metrics - pub static ref TWITTER_API_CALLS_TOTAL: IntCounterVec = IntCounterVec::new( - Opts::new("twitter_api_calls_total", "Total number of Twitter API calls"), - &["operation"] - ) - .unwrap(); - pub static ref TWITTER_API_CALL_DURATION: HistogramVec = HistogramVec::new( - HistogramOpts::new("twitter_api_call_duration_seconds", "Twitter API call duration in seconds") - .buckets(vec![0.1, 0.25, 0.5, 1.0, 2.5, 5.0, 10.0, 30.0]), - &["operation"] - ) - .unwrap(); - pub static ref TWITTER_TWEETS_PULLED_TOTAL: IntCounterVec = IntCounterVec::new( - Opts::new("twitter_tweets_pulled_total", "Total number of tweets pulled from Twitter API"), - &["operation"] - ) - .unwrap(); - pub static ref TWITTER_API_ERRORS_TOTAL: IntCounterVec = IntCounterVec::new( - Opts::new("twitter_api_errors_total", "Total number of Twitter API errors"), - &["operation", "error_type"] - ) - .unwrap(); - pub static ref TWITTER_TWEETS_PER_CALL: HistogramVec = HistogramVec::new( - HistogramOpts::new("twitter_tweets_per_call", "Number of tweets returned per API call") - .buckets(vec![0.0, 1.0, 5.0, 10.0, 25.0, 50.0, 100.0]), - &["operation"] - ) - .unwrap(); } impl Default for Metrics { @@ -111,15 +81,6 @@ impl Metrics { registry.register(Box::new(HTTP_RESPONSE_SIZE_BYTES.clone())).unwrap(); registry.register(Box::new(HTTP_ERRORS_TOTAL.clone())).unwrap(); - // Register Twitter API metrics - registry.register(Box::new(TWITTER_API_CALLS_TOTAL.clone())).unwrap(); - registry.register(Box::new(TWITTER_API_CALL_DURATION.clone())).unwrap(); - registry - .register(Box::new(TWITTER_TWEETS_PULLED_TOTAL.clone())) - .unwrap(); - registry.register(Box::new(TWITTER_API_ERRORS_TOTAL.clone())).unwrap(); - registry.register(Box::new(TWITTER_TWEETS_PER_CALL.clone())).unwrap(); - Self { registry: Arc::new(registry), } @@ -217,85 +178,6 @@ pub async fn track_metrics(req: Request, next: Next) -> Response { response } -fn extract_sdk_error_type(err: &SdkError) -> &'static str { - match err { - SdkError::Api { status, .. } => { - // Categorize based on HTTP status code - // Match specific status codes first, then ranges - match *status { - 400 => "bad_request", - 401 => "unauthorized", - 403 => "forbidden", - 404 => "not_found", - 429 => "rate_limit", - 500..=599 => "server_error", - 402 | 405..=428 | 430..=499 => "client_error", - _ => "twitter_api_error", - } - } - _ => "sdk_error", - } -} - -/// Track Twitter API call metrics with error type detection -/// -/// This function should be called around Twitter API calls to track: -/// - API call duration -/// - Number of tweets pulled -/// - Errors with specific error types extracted from SdkError -/// -/// # Arguments -/// * `operation` - The type of operation (e.g., "search_recent", "tweets_get_many") -/// * `f` - The async function that makes the Twitter API call -/// -/// # Returns -/// The result of the API call, with metrics automatically tracked -pub async fn track_twitter_api_call(operation: &str, f: F) -> Result -where - F: std::future::Future>, -{ - let start = Instant::now(); - - // Track API call attempt - TWITTER_API_CALLS_TOTAL.with_label_values(&[operation]).inc(); - - let result = f.await; - - // Record duration - let duration = start.elapsed().as_secs_f64(); - TWITTER_API_CALL_DURATION - .with_label_values(&[operation]) - .observe(duration); - - // Track errors with specific error types - if let Err(ref err) = result { - let error_type = extract_sdk_error_type(err); - TWITTER_API_ERRORS_TOTAL - .with_label_values(&[operation, error_type]) - .inc(); - } - - result -} - -/// Track tweets pulled from a Twitter API call -/// -/// Call this after successfully getting tweets from the API to track: -/// - Total tweets pulled -/// - Tweets per call distribution -/// -/// # Arguments -/// * `operation` - The type of operation (e.g., "search_recent", "tweets_get_many") -/// * `tweet_count` - Number of tweets returned by the API call -pub fn track_tweets_pulled(operation: &str, tweet_count: usize) { - TWITTER_TWEETS_PULLED_TOTAL - .with_label_values(&[operation]) - .inc_by(tweet_count as u64); - TWITTER_TWEETS_PER_CALL - .with_label_values(&[operation]) - .observe(tweet_count as f64); -} - pub async fn metrics_handler(State(state): State) -> impl IntoResponse { let encoder = TextEncoder::new(); let metric_families = state.metrics.registry.gather(); diff --git a/src/models/mod.rs b/src/models/mod.rs index 4b5a4cf..3b2b358 100644 --- a/src/models/mod.rs +++ b/src/models/mod.rs @@ -15,4 +15,3 @@ pub mod raid_quest; pub mod referrals; pub mod relevant_tweet; pub mod tweet_author; -pub mod tweet_pull_usage; diff --git a/src/models/relevant_tweet.rs b/src/models/relevant_tweet.rs index 6ee4261..6ed0f9b 100644 --- a/src/models/relevant_tweet.rs +++ b/src/models/relevant_tweet.rs @@ -1,5 +1,4 @@ use chrono::{DateTime, Utc}; -use rusx::resources::tweet::{Tweet as TwitterTweet, TweetPublicMetrics}; use serde::{Deserialize, Serialize}; use sqlx::{postgres::PgRow, FromRow, Row}; @@ -88,24 +87,3 @@ pub struct NewTweetPayload { pub like_count: i32, pub created_at: DateTime, } - -impl NewTweetPayload { - pub fn new(tweet: TwitterTweet) -> Self { - let public_metrics = tweet - .public_metrics - .ok_or_else(|| TweetPublicMetrics { ..Default::default() }) - .unwrap(); - let created_at = tweet.created_at.ok_or_else(|| chrono::Utc::now().to_rfc3339()).unwrap(); - - NewTweetPayload { - id: tweet.id, - author_id: tweet.author_id.unwrap(), - text: tweet.text, - impression_count: public_metrics.impression_count as i32, - like_count: public_metrics.like_count as i32, - retweet_count: public_metrics.retweet_count as i32, - reply_count: public_metrics.reply_count as i32, - created_at: DateTime::parse_from_rfc3339(&created_at).unwrap().with_timezone(&Utc), - } - } -} diff --git a/src/models/tweet_pull_usage.rs b/src/models/tweet_pull_usage.rs deleted file mode 100644 index 6d247a7..0000000 --- a/src/models/tweet_pull_usage.rs +++ /dev/null @@ -1,10 +0,0 @@ -use chrono::{DateTime, Utc}; -use serde::{Deserialize, Serialize}; -use sqlx::FromRow; - -#[derive(Debug, Clone, Serialize, Deserialize, FromRow)] -pub struct TweetPullUsage { - pub period: String, - pub tweet_count: i32, - pub updated_at: DateTime, -} diff --git a/src/repositories/mod.rs b/src/repositories/mod.rs index d5e7ee9..48ec298 100644 --- a/src/repositories/mod.rs +++ b/src/repositories/mod.rs @@ -10,7 +10,6 @@ pub mod raid_quest; pub mod referral; pub mod relevant_tweet; pub mod tweet_author; -pub mod tweet_pull_usage; pub trait QueryBuilderExt { fn push_condition(&mut self, sql: &str, where_started: &mut bool); diff --git a/src/repositories/relevant_tweet.rs b/src/repositories/relevant_tweet.rs index ab060fd..19ab1af 100644 --- a/src/repositories/relevant_tweet.rs +++ b/src/repositories/relevant_tweet.rs @@ -1,4 +1,4 @@ -use sqlx::{PgPool, Postgres, QueryBuilder, Row}; +use sqlx::{PgPool, Postgres, QueryBuilder}; use crate::{ db_persistence::DbError, @@ -90,14 +90,6 @@ impl RelevantTweetRepository { Ok(count) } - pub async fn get_newest_tweet_id(&self) -> Result, DbError> { - let row = sqlx::query("SELECT id FROM relevant_tweets WHERE created_at >= NOW() - INTERVAL '7 days' ORDER BY created_at DESC LIMIT 1") - .fetch_optional(&self.pool) - .await?; - - Ok(row.map(|r| r.get("id"))) - } - /// Find all tweets with author details joined pub async fn find_all_with_authors( &self, @@ -229,7 +221,7 @@ mod tests { utils::test_db::reset_database, Config, }; - use chrono::{Duration, Utc}; + use chrono::Utc; use sqlx::PgPool; // --- Helpers to create dummy data --- @@ -312,39 +304,4 @@ mod tests { "Should update existing record on conflict" ); } - - #[tokio::test] - async fn test_get_newest_tweet_id_returns_none_when_no_recent_tweets() { - let (repo, author_repo) = setup_test_repository().await; - let author_id = "author_old"; - seed_author(&author_repo, author_id, "stale_user").await; - - let mut old_tweet = create_payload("tweet_old", author_id, "stale tweet"); - old_tweet.created_at = Utc::now() - Duration::days(8); - repo.upsert_many(&vec![old_tweet]).await.unwrap(); - - let newest_tweet_id = repo.get_newest_tweet_id().await.unwrap(); - assert_eq!( - newest_tweet_id, None, - "Should return None when no tweet is within the last 7 days" - ); - } - - #[tokio::test] - async fn test_get_newest_tweet_id_returns_some_when_recent_tweet_exists() { - let (repo, author_repo) = setup_test_repository().await; - let author_id = "author_recent"; - seed_author(&author_repo, author_id, "active_user").await; - - let mut recent_tweet = create_payload("tweet_recent", author_id, "recent tweet"); - recent_tweet.created_at = Utc::now() - Duration::days(2); - repo.upsert_many(&vec![recent_tweet]).await.unwrap(); - - let newest_tweet_id = repo.get_newest_tweet_id().await.unwrap(); - assert_eq!( - newest_tweet_id, - Some("tweet_recent".to_string()), - "Should return the newest tweet id when at least one tweet is within the last 7 days" - ); - } } diff --git a/src/repositories/tweet_author.rs b/src/repositories/tweet_author.rs index 19ac549..9ef5c3a 100644 --- a/src/repositories/tweet_author.rs +++ b/src/repositories/tweet_author.rs @@ -107,14 +107,6 @@ impl TweetAuthorRepository { Ok(authors) } - pub async fn get_whitelist(&self) -> Result, DbError> { - let usernames = sqlx::query_scalar::<_, String>("SELECT username FROM tweet_authors WHERE is_ignored = false") - .fetch_all(&self.pool) - .await?; - - Ok(usernames) - } - pub async fn set_ignore_status(&self, id: &str, status: bool) -> Result<(), DbError> { sqlx::query("UPDATE tweet_authors SET is_ignored = $1 WHERE id = $2") .bind(status) diff --git a/src/repositories/tweet_pull_usage.rs b/src/repositories/tweet_pull_usage.rs deleted file mode 100644 index d1adcd8..0000000 --- a/src/repositories/tweet_pull_usage.rs +++ /dev/null @@ -1,265 +0,0 @@ -use crate::db_persistence::DbError; -use crate::models::tweet_pull_usage::TweetPullUsage; -use chrono::{Datelike, NaiveDate, Utc}; -use sqlx::PgPool; - -#[derive(Debug, Clone)] -pub struct TweetPullUsageRepository { - pool: PgPool, -} - -impl TweetPullUsageRepository { - pub fn new(pool: PgPool) -> Self { - Self { pool } - } - - fn get_current_period(reset_day: u32) -> String { - Self::calculate_period_for_date(Utc::now(), reset_day) - } - - /// Calculates the billing period string (YYYY-MM) based on the given date and a reset day. - /// - /// The logic handles edge cases where the reset day (e.g., 31st) doesn't exist in the current month. - /// In such cases, the reset day effectively becomes the last day of that month. - /// - /// Note on Month Label Shifting: - /// If `reset_day` is 31, the "2023-01" cycle starts on Jan 31st and ends on Feb 27th. - /// The "2023-02" cycle starts on Feb 28th and ends on Mar 30th. - /// This means most of the usage for the "2023-01" period actually occurs in February. - /// This is consistent with most billing systems that reset on the last day of the month - /// for shorter months, but it means the month label refers to the START of the billing cycle. - fn calculate_period_for_date(date: chrono::DateTime, reset_day: u32) -> String { - let current_year = date.year(); - let current_month = date.month(); - let current_day = date.day(); - - // 1. Determine the effective reset day for the CURRENT month. - // If `reset_day` exceeds the days in current month, cap it at the last day of the month. - let days_in_current_month = get_days_in_month(current_year, current_month); - let effective_reset_day_current_month = std::cmp::min(reset_day, days_in_current_month); - - // 2. Compare current day with the effective reset day. - if current_day >= effective_reset_day_current_month { - // We are in the cycle starting this month. - format!("{}-{:02}", current_year, current_month) - } else { - // We are in the cycle that started last month. - let (prev_year, prev_month) = if current_month == 1 { - (current_year - 1, 12) - } else { - (current_year, current_month - 1) - }; - format!("{}-{:02}", prev_year, prev_month) - } - } - - pub async fn increment_usage(&self, amount: i32, reset_day: u32) -> Result { - let period = Self::get_current_period(reset_day); - self.increment_usage_for_period(amount, &period).await - } - - /// Internal helper to increment usage for a specific period string. - async fn increment_usage_for_period(&self, amount: i32, period: &str) -> Result { - let usage = sqlx::query_as::<_, TweetPullUsage>( - "INSERT INTO tweet_pull_usage (period, tweet_count) - VALUES ($1, $2) - ON CONFLICT (period) DO UPDATE - SET tweet_count = tweet_pull_usage.tweet_count + EXCLUDED.tweet_count - RETURNING *", - ) - .bind(period) - .bind(amount) - .fetch_one(&self.pool) - .await - .map_err(DbError::Database)?; - - Ok(usage) - } -} - -/// Helper to get the number of days in a given month/year. -fn get_days_in_month(year: i32, month: u32) -> u32 { - // If month is December (12), next month is Jan (1) of next year. - // Otherwise, just next month of same year. - let (next_year, next_month) = if month == 12 { (year + 1, 1) } else { (year, month + 1) }; - - // The '0th' day of the next month is the last day of the current month. - // We get the date of the 1st of the next month, subtract 1 day. - NaiveDate::from_ymd_opt(next_year, next_month, 1) - .unwrap() - .pred_opt() - .unwrap() - .day() -} - -#[cfg(test)] -mod tests { - use super::*; - use crate::utils::test_app_state::create_test_app_state; - use crate::utils::test_db::reset_database; - use chrono::{TimeZone, Utc}; - use sqlx::{Pool, Postgres}; - - async fn get_current_usage(pool: &Pool, reset_day: u32) -> Result { - let period = TweetPullUsageRepository::get_current_period(reset_day); - get_usage_for_period(pool, &period).await - } - - /// Internal helper to get usage for a specific period string. - async fn get_usage_for_period(pool: &Pool, period: &str) -> Result { - let usage = sqlx::query_as::<_, TweetPullUsage>( - "INSERT INTO tweet_pull_usage (period, tweet_count) - VALUES ($1, 0) - ON CONFLICT (period) DO UPDATE SET period = EXCLUDED.period - RETURNING *", - ) - .bind(period) - .fetch_one(pool) - .await - .map_err(DbError::Database)?; - - Ok(usage) - } - - #[tokio::test] - async fn test_get_current_usage_integration() { - let state = create_test_app_state().await; - reset_database(&state.db.pool).await; - - let reset_day = 1; - - // 1. Initial call should create a record with 0 - let usage = get_current_usage(&state.db.pool, reset_day).await.unwrap(); - assert_eq!(usage.tweet_count, 0); - - // 2. Subsequent call should return the same record - let usage2 = get_current_usage(&state.db.pool, reset_day).await.unwrap(); - assert_eq!(usage2.tweet_count, 0); - assert_eq!(usage.period, usage2.period); - } - - #[tokio::test] - async fn test_increment_usage_integration() { - let state = create_test_app_state().await; - reset_database(&state.db.pool).await; - - let repo = &state.db.tweet_pull_usage; - let reset_day = 1; - - // 1. Increment from zero - let usage = repo.increment_usage(10, reset_day).await.unwrap(); - assert_eq!(usage.tweet_count, 10); - - // 2. Increment again - let usage2 = repo.increment_usage(5, reset_day).await.unwrap(); - assert_eq!(usage2.tweet_count, 15); - } - - #[tokio::test] - async fn test_transition_between_months_integration() { - let state = create_test_app_state().await; - reset_database(&state.db.pool).await; - - let repo = &state.db.tweet_pull_usage; - - // 1. Increment for Month A - let period_a = "2023-01"; - repo.increment_usage_for_period(100, period_a).await.unwrap(); - - // 2. Increment for Month B - let period_b = "2023-02"; - repo.increment_usage_for_period(50, period_b).await.unwrap(); - - // 3. Verify they are separate - let usage_a = get_usage_for_period(&state.db.pool, period_a).await.unwrap(); - let usage_b = get_usage_for_period(&state.db.pool, period_b).await.unwrap(); - - assert_eq!(usage_a.tweet_count, 100); - assert_eq!(usage_b.tweet_count, 50); - assert_ne!(usage_a.period, usage_b.period); - } - - #[test] - fn test_period_standard_reset() { - // Reset on 7th. Current is Jan 5th. Should be Dec cycle. - let date = Utc.with_ymd_and_hms(2023, 1, 5, 0, 0, 0).unwrap(); - assert_eq!(TweetPullUsageRepository::calculate_period_for_date(date, 7), "2022-12"); - - // Reset on 7th. Current is Jan 7th. Should be Jan cycle. - let date = Utc.with_ymd_and_hms(2023, 1, 7, 0, 0, 0).unwrap(); - assert_eq!(TweetPullUsageRepository::calculate_period_for_date(date, 7), "2023-01"); - } - - #[test] - fn test_period_end_of_month_reset_short_feb() { - // Reset on 30th. - // Feb 2023 has 28 days. - - // Date: Feb 27th. - // Effective reset for Feb is 28th (min(30, 28)). - // 27 < 28 -> Previous cycle (Jan). - let date = Utc.with_ymd_and_hms(2023, 2, 27, 0, 0, 0).unwrap(); - assert_eq!(TweetPullUsageRepository::calculate_period_for_date(date, 30), "2023-01"); - - // Date: Feb 28th. - // 28 >= 28 -> Current cycle (Feb). - let date = Utc.with_ymd_and_hms(2023, 2, 28, 0, 0, 0).unwrap(); - assert_eq!(TweetPullUsageRepository::calculate_period_for_date(date, 30), "2023-02"); - } - - #[test] - fn test_period_leap_year() { - // Reset on 30th. - // Feb 2024 has 29 days. - - // Date: Feb 28th. - // Effective reset for Feb is 29th (min(30, 29)). - // 28 < 29 -> Previous cycle (Jan). - let date = Utc.with_ymd_and_hms(2024, 2, 28, 0, 0, 0).unwrap(); - assert_eq!(TweetPullUsageRepository::calculate_period_for_date(date, 30), "2024-01"); - - // Date: Feb 29th. - // 29 >= 29 -> Current cycle (Feb). - let date = Utc.with_ymd_and_hms(2024, 2, 29, 0, 0, 0).unwrap(); - assert_eq!(TweetPullUsageRepository::calculate_period_for_date(date, 30), "2024-02"); - } - - #[test] - fn test_period_new_year() { - // Reset on 31st. Current is Jan 1st 2024. - // Effective reset for Jan is 31st. - // 1 < 31 -> Previous cycle (Dec 2023). - let date = Utc.with_ymd_and_hms(2024, 1, 1, 0, 0, 0).unwrap(); - assert_eq!(TweetPullUsageRepository::calculate_period_for_date(date, 31), "2023-12"); - } - - #[test] - fn test_period_reset_day_31_feb_edge_cases() { - // Reset on 31st. - - // Jan 31st -> Starts Jan cycle. - let date = Utc.with_ymd_and_hms(2023, 1, 31, 0, 0, 0).unwrap(); - assert_eq!(TweetPullUsageRepository::calculate_period_for_date(date, 31), "2023-01"); - - // Feb 27th -> Still in Jan cycle (started Jan 31). - let date = Utc.with_ymd_and_hms(2023, 2, 27, 0, 0, 0).unwrap(); - assert_eq!(TweetPullUsageRepository::calculate_period_for_date(date, 31), "2023-01"); - - // Feb 28th -> Starts Feb cycle because Feb only has 28 days. - // The reset is "capped" at the last day of the month. - let date = Utc.with_ymd_and_hms(2023, 2, 28, 0, 0, 0).unwrap(); - assert_eq!(TweetPullUsageRepository::calculate_period_for_date(date, 31), "2023-02"); - - // Mar 1st -> Still in Feb cycle (started Feb 28). - let date = Utc.with_ymd_and_hms(2023, 3, 1, 0, 0, 0).unwrap(); - assert_eq!(TweetPullUsageRepository::calculate_period_for_date(date, 31), "2023-02"); - - // Mar 30th -> Still in Feb cycle. - let date = Utc.with_ymd_and_hms(2023, 3, 30, 0, 0, 0).unwrap(); - assert_eq!(TweetPullUsageRepository::calculate_period_for_date(date, 31), "2023-02"); - - // Mar 31st -> Starts Mar cycle. - let date = Utc.with_ymd_and_hms(2023, 3, 31, 0, 0, 0).unwrap(); - assert_eq!(TweetPullUsageRepository::calculate_period_for_date(date, 31), "2023-03"); - } -} diff --git a/src/services/alert_service.rs b/src/services/alert_service.rs deleted file mode 100644 index 7691027..0000000 --- a/src/services/alert_service.rs +++ /dev/null @@ -1,91 +0,0 @@ -use crate::repositories::tweet_pull_usage::TweetPullUsageRepository; -use crate::{AppResult, Config}; -use reqwest::Client; -use serde::Serialize; - -#[derive(Debug, Clone)] -pub struct AlertService { - client: Client, - webhook_url: String, - config: Config, - usage_repo: TweetPullUsageRepository, -} - -#[derive(Serialize)] -struct WebhookPayload<'a> { - text: &'a str, -} - -impl AlertService { - pub fn new(config: Config, usage_repo: TweetPullUsageRepository) -> Self { - Self { - client: Client::new(), - webhook_url: config.alert.webhook_url.clone(), - config, - usage_repo, - } - } - - /// Increments Twitter API usage and sends an alert if the threshold is reached. - pub async fn track_and_alert_usage(&self, tweets_pulled: i32) -> AppResult<()> { - if tweets_pulled <= 0 { - return Ok(()); - } - - match self - .usage_repo - .increment_usage(tweets_pulled, self.config.tweet_sync.reset_day) - .await - { - Ok(usage) => { - let current_total = usage.tweet_count as u32; - if current_total >= self.config.tweet_sync.alert_threshold { - if let Err(e) = self - .send_twitter_limit_alert(current_total, self.config.tweet_sync.monthly_limit) - .await - { - tracing::error!("Failed to send Twitter limit alert: {:?}", e); - } - } - } - Err(e) => { - tracing::error!("Failed to increment Twitter API usage: {:?}", e); - } - } - - Ok(()) - } - - /// Sends an alert when the Twitter API tweet pull limit is nearing its threshold. - pub async fn send_twitter_limit_alert(&self, current_count: u32, limit: u32) -> AppResult<()> { - let message = format!( - "Twitter API Limit Warning: Current usage {} tweets pulled, Plan limit {} tweets ({:.1}% used).", - current_count, - limit, - (current_count as f32 / limit as f32) * 100.0 - ); - - self.send_webhook_alert(&message).await - } - - pub async fn send_webhook_alert(&self, text: &str) -> AppResult<()> { - let payload = WebhookPayload { text }; - - let response = self.client.post(&self.webhook_url).json(&payload).send().await; - - match response { - Ok(res) if res.status().is_success() => Ok(()), - Ok(res) => { - let status = res.status().as_u16(); - let body = res.text().await.unwrap_or_default(); - // Reusing Telegram error variant for now or we could add a new one. - // Given the instructions, let's keep it simple or add a Generic error. - Err(crate::AppError::Server(format!( - "Webhook alert failed with status {}: {}", - status, body - ))) - } - Err(e) => Err(crate::AppError::Server(format!("Webhook alert request failed: {}", e))), - } - } -} diff --git a/src/services/mod.rs b/src/services/mod.rs index 558ccc5..99dfa39 100644 --- a/src/services/mod.rs +++ b/src/services/mod.rs @@ -1,8 +1,5 @@ -pub mod alert_service; pub mod exchange_rate_service; pub mod graphql_client; pub mod risk_checker_service; pub mod signature_service; -pub mod telegram_service; -pub mod tweet_synchronizer_service; pub mod wallet_config_service; diff --git a/src/services/telegram_service.rs b/src/services/telegram_service.rs deleted file mode 100644 index f214e6b..0000000 --- a/src/services/telegram_service.rs +++ /dev/null @@ -1,91 +0,0 @@ -use reqwest::{Client, StatusCode}; -use serde::Serialize; - -use crate::{config::TelegramBotConfig, AppError, AppResult}; - -#[derive(Clone)] -pub struct TelegramService { - client: Client, - base_url: String, - default_chat_id: String, - default_message_thread_id: Option, -} - -#[derive(Serialize)] -struct MessagePayload<'a> { - chat_id: &'a str, - #[serde(skip_serializing_if = "Option::is_none")] - message_thread_id: Option<&'a str>, - text: &'a str, - parse_mode: &'a str, - disable_web_page_preview: bool, -} - -impl TelegramService { - pub fn escape_markdown_v2(text: &str) -> String { - text.replace("_", "\\_") - // .replace("*", "\\*") // We want to allow bolding with ** - .replace("[", "\\[") - .replace("]", "\\]") - .replace("(", "\\(") - .replace(")", "\\)") - .replace("~", "\\~") - .replace("`", "\\`") - .replace(">", "\\>") - .replace("#", "\\#") - .replace("+", "\\+") - .replace("-", "\\-") - .replace("=", "\\=") - .replace("|", "\\|") - .replace("{", "\\{") - .replace("}", "\\}") - .replace(".", "\\.") - .replace("!", "\\!") - } - - pub fn new(config: TelegramBotConfig) -> Self { - Self { - client: Client::new(), - base_url: format!("{}/bot{}", config.base_url, config.token), - default_chat_id: config.chat_id, - default_message_thread_id: config.message_thread_id, - } - } - - pub async fn send_message(&self, text: &str) -> AppResult<()> { - self.send(&self.default_chat_id, self.default_message_thread_id.as_deref(), text) - .await - } - - async fn send(&self, chat_id: &str, message_thread_id: Option<&str>, text: &str) -> AppResult<()> { - let url = format!("{}/sendMessage", self.base_url); - - let payload = MessagePayload { - chat_id, - message_thread_id, - text, - parse_mode: "MarkdownV2", // or "HTML" - disable_web_page_preview: true, - }; - - let response = self.client.post(&url).json(&payload).send().await; - match response { - Ok(response) => { - if !response.status().is_success() { - let status = response.status().as_u16(); - let body = response.text().await.unwrap_or_default(); - - return Err(AppError::Telegram(status, body)); - } - - Ok(()) - } - Err(err) => { - let status = err.status().unwrap_or(StatusCode::INTERNAL_SERVER_ERROR).as_u16(); - let body = err.to_string(); - - Err(AppError::Telegram(status, body)) - } - } - } -} diff --git a/src/services/tweet_synchronizer_service.rs b/src/services/tweet_synchronizer_service.rs deleted file mode 100644 index 34d096c..0000000 --- a/src/services/tweet_synchronizer_service.rs +++ /dev/null @@ -1,652 +0,0 @@ -use std::{collections::HashMap, sync::Arc}; - -use rusx::{ - resources::{ - search::{SearchParams, SearchSortOrder}, - tweet::Tweet, - TweetExpansion, TweetField, TwitterApiResponse, UserField, - }, - TwitterGateway, -}; - -use crate::{ - db_persistence::DbPersistence, - metrics::{track_tweets_pulled, track_twitter_api_call}, - models::{relevant_tweet::NewTweetPayload, tweet_author::NewAuthorPayload}, - services::{alert_service::AlertService, telegram_service::TelegramService}, - utils::x_url::build_x_status_url, - AppError, AppResult, Config, -}; - -#[derive(Clone)] -pub struct TweetSynchronizerService { - db: Arc, - twitter_gateway: Arc, - telegram_service: Arc, - alert_service: Arc, - config: Arc, -} - -impl TweetSynchronizerService { - async fn process_tweet_authors( - &self, - response: &TwitterApiResponse>, - ) -> Result, AppError> { - let Some(includes) = &response.includes else { - tracing::info!("No authors found in includes."); - return Ok(Default::default()); - }; - let Some(authors) = &includes.users else { - tracing::info!("No users found in includes."); - return Ok(Default::default()); - }; - - let authors_found_count = authors.len(); - tracing::info!("Found {} new authors. Saving...", authors_found_count); - - let new_authors: Vec = authors - .iter() - .map(|author| NewAuthorPayload::new(author.clone())) - .collect(); - - if !new_authors.is_empty() { - self.db.tweet_authors.upsert_many(&new_authors).await?; - tracing::info!("Success saving {} new authors.", authors_found_count); - } - - Ok(new_authors) - } - - async fn process_relevant_tweets( - &self, - response: &TwitterApiResponse>, - ) -> Result, AppError> { - let Some(tweets) = &response.data else { - tracing::info!("No new relevant tweets found."); - return Ok(Default::default()); - }; - - let tweets_found_count = tweets.len(); - tracing::info!("Found {} new relevant tweets. Saving...", tweets_found_count); - - let new_relevant_tweets: Vec = - tweets.iter().map(|tweet| NewTweetPayload::new(tweet.clone())).collect(); - - if !new_relevant_tweets.is_empty() { - self.db.relevant_tweets.upsert_many(&new_relevant_tweets).await?; - tracing::info!("Success saving {} new relevant tweets.", tweets_found_count); - } - - Ok(new_relevant_tweets) - } - - async fn process_sending_raid_targets( - &self, - tweet_authors: &[NewAuthorPayload], - relevant_tweets: &[NewTweetPayload], - ) -> Result<(), AppError> { - if relevant_tweets.is_empty() { - return Ok(()); - } - - let active_raid = self.db.raid_quests.find_active().await?; - - if active_raid.is_some() { - tracing::info!("Active raid quests found, sending raid targets..."); - - let author_lookup: HashMap<&String, &String> = tweet_authors.iter().map(|a| (&a.id, &a.username)).collect(); - let telegram_service = self.telegram_service.clone(); - let tweets_to_process = relevant_tweets.to_vec(); - let mut messages: Vec = Vec::with_capacity(tweets_to_process.len()); - - for tweet in tweets_to_process.iter().rev() { - let author_name = match author_lookup.get(&tweet.author_id) { - Some(name) => name, - None => { - tracing::warn!( - "Author ID {} not found in lookup. Skipping notification.", - tweet.author_id - ); - "Unknown" - } - }; - - let link = build_x_status_url(author_name, &tweet.id); - - let escaped_link = TelegramService::escape_markdown_v2(&link); - let escaped_author = TelegramService::escape_markdown_v2(author_name); - let escaped_text = TelegramService::escape_markdown_v2(&tweet.text); - let escaped_posted_at = TelegramService::escape_markdown_v2(&tweet.created_at.to_rfc3339()); - - let tg_message = format!( - "*Raid Target Found\\!*\n\n*Link*: {}\n\n*Author*: {}\n\n*Text*: {}\n\n*Impressions*: {}\n\n*Posted At*: {}", - escaped_link, - escaped_author, - escaped_text, - tweet.impression_count, - escaped_posted_at - ); - messages.push(tg_message); - } - - tokio::spawn(async move { - // Telegram Limit: ~30 messages per second. - // We set interval to 50ms (~20 msgs/sec) to be safe. - let mut rate_limiter = tokio::time::interval(tokio::time::Duration::from_millis(50)); - - for msg in messages { - rate_limiter.tick().await; - - if let Err(e) = telegram_service.send_message(&msg).await { - tracing::error!("Failed to send raid notification: {:?}", e); - } - } - }); - } - - Ok(()) - } - - pub fn new( - db: Arc, - twitter_gateway: Arc, - telegram_service: Arc, - alert_service: Arc, - config: Arc, - ) -> Self { - Self { - db, - twitter_gateway, - telegram_service, - alert_service, - config, - } - } - - pub fn spawn_tweet_synchronizer(&self) -> tokio::task::JoinHandle> { - let service = self.clone(); - - tokio::spawn(async move { - let mut interval = tokio::time::interval(service.config.get_tweet_sync_interval()); - - loop { - interval.tick().await; - tracing::info!("🔄 Background Worker: Starting Twitter Sync..."); - - match service.sync_relevant_tweets().await { - Ok(_) => tracing::info!("✅ Sync Complete. Relevant tweets synced."), - Err(e) => tracing::error!("❌ Sync Failed: {:?}", e), - } - } - }) - } - - pub async fn sync_relevant_tweets(&self) -> Result<(), AppError> { - let last_id = self.db.relevant_tweets.get_newest_tweet_id().await?; - let whitelist = self.db.tweet_authors.get_whitelist().await?; - - let whitelist_queries = - SearchParams::build_batched_whitelist_queries(&whitelist, Some(&self.config.tweet_sync.keywords)); - - for query in whitelist_queries { - let mut params = SearchParams::new(query); - params.max_results = Some(100); - - params.sort_order = Some(SearchSortOrder::Recency); - - params.tweet_fields = Some(vec![ - TweetField::PublicMetrics, - TweetField::CreatedAt, - TweetField::AuthorId, - ]); - params.user_fields = Some(vec![ - UserField::Username, - UserField::Name, - UserField::Id, - UserField::PublicMetrics, - ]); - params.expansions = Some(vec![TweetExpansion::AuthorId]); - - if let Some(id) = last_id.clone() { - params.since_id = Some(id.clone()); - tracing::info!("Syncing tweets since ID: {}", id); - } else { - tracing::info!("No history found, performing full 7-day fetch."); - } - - // Track Twitter API call with metrics - let response = track_twitter_api_call("search_recent", async { - self.twitter_gateway.search().recent(params).await - }) - .await?; - - let tweet_authors = self.process_tweet_authors(&response).await?; - let relevant_tweets = self.process_relevant_tweets(&response).await?; - - // Track Twitter API usage (for alerting) - let tweets_pulled = relevant_tweets.len(); - self.alert_service.track_and_alert_usage(tweets_pulled as i32).await?; - - // Track metrics for tweets pulled - track_tweets_pulled("search_recent", tweets_pulled); - - // We might get duplicate because X will return the tweet including the since_id not just the new ones. - let new_tweets: Vec = relevant_tweets - .iter() - .filter(|t| Some(&t.id) != last_id.as_ref()) - .cloned() - .collect(); - - self.process_sending_raid_targets(&tweet_authors, &new_tweets).await?; - } - - Ok(()) - } -} - -#[cfg(test)] -mod tests { - // NOTE: These tests share the test DB; run with `cargo test ... -- --test-threads=1` (see scripts/run_non_chain_tests.sh). - use super::*; - use crate::config::{Config, TelegramBotConfig}; - use crate::models::raid_quest::CreateRaidQuest; - use crate::utils::test_db::reset_database; - use mockall::predicate::*; - use rusx::{ - resources::{ - search::SearchApi, - tweet::{Tweet, TweetPublicMetrics}, - user::{User, UserPublicMetrics}, - Includes, TwitterApiResponse, - }, - MockSearchApi, MockTwitterGateway, - }; - use sqlx::PgPool; - use std::sync::Arc; - use wiremock::matchers::{method, path}; - use wiremock::{Mock, MockServer, ResponseTemplate}; - - // ------------------------------------------------------------------------- - // Test Helpers - // ------------------------------------------------------------------------- - - async fn setup_deps() -> ( - Arc, - MockServer, - Arc, - Arc, - Arc, - ) { - // A. Setup DB - let config = Config::load_test_env().expect("Failed to load test config"); - let pool = PgPool::connect(config.get_database_url()).await.unwrap(); - reset_database(&pool).await; - let db = Arc::new(DbPersistence::new(config.get_database_url()).await.unwrap()); - - // Seed an author for the whitelist so sync loop runs - db.tweet_authors - .upsert_many(&vec![crate::models::tweet_author::NewAuthorPayload { - id: "u1".to_string(), - name: "User One".to_string(), - username: "user_one".to_string(), - followers_count: 1000, - following_count: 100, - tweet_count: 50, - listed_count: 5, - like_count: 0, - media_count: 0, - is_ignored: Some(false), - }]) - .await - .unwrap(); - - // B. Setup Telegram Mock Server - let mock_server = MockServer::start().await; - let telegram_config = TelegramBotConfig { - base_url: mock_server.uri(), - token: "123456".to_string(), - chat_id: config.tg_bot.chat_id.clone(), - message_thread_id: config.tg_bot.message_thread_id.clone(), - }; - let telegram_service = Arc::new(TelegramService::new(telegram_config)); - - // C. Config - let app_config = Arc::new(config.clone()); - - // D. Alert Service - let alert_service = Arc::new(AlertService::new(config.clone(), db.tweet_pull_usage.clone())); - - (db, mock_server, telegram_service, alert_service, app_config) - } - - fn create_mock_tweet(id: &str, author_id: &str) -> Tweet { - Tweet { - id: id.to_string(), - text: "Hello World".to_string(), - author_id: Some(author_id.to_string()), - created_at: Some(chrono::Utc::now().to_rfc3339()), - in_reply_to_user_id: None, - referenced_tweets: None, - public_metrics: Some(TweetPublicMetrics { - impression_count: 100, - like_count: 10, - reply_count: 5, - retweet_count: 2, - ..Default::default() - }), - } - } - - fn create_mock_user(id: &str, username: &str) -> User { - User { - id: id.to_string(), - name: "Test User".to_string(), - username: username.to_string(), - description: Some("Quantus Network".to_string()), - public_metrics: Some(UserPublicMetrics { - followers_count: 1000, - following_count: 100, - tweet_count: 50, - listed_count: 5, - media_count: Some(0), - like_count: Some(0), - }), - } - } - - // ------------------------------------------------------------------------- - // Tests - // ------------------------------------------------------------------------- - - #[tokio::test] - async fn test_sync_saves_data_no_raid_notification() { - let (db, _mock_tg, telegram_service, alert_service, config) = setup_deps().await; - - // --- Setup DB with existing watched tweet authors --- - let dummy_author = crate::models::tweet_author::NewAuthorPayload { - id: "old_u".to_string(), - name: "Old".to_string(), - username: "old".to_string(), - followers_count: 0, - following_count: 0, - tweet_count: 0, - listed_count: 0, - like_count: 0, - media_count: 0, - is_ignored: Some(false), - }; - - db.tweet_authors.upsert(&dummy_author).await.unwrap(); - - // --- Setup Mocks --- - let mut mock_gateway = MockTwitterGateway::new(); - let mut mock_search = MockSearchApi::new(); - - // Expect search().recent() to be called - let mock_data = create_mock_tweet("t1", &dummy_author.id); - let mock_users = create_mock_user(&dummy_author.id, &dummy_author.username); - - mock_search - .expect_recent() - .times(1) // Expect 1 batch (based on whitelist loop) - .returning(move |_| { - Ok(TwitterApiResponse::> { - data: Some(vec![mock_data.clone()]), - includes: Some(Includes { - users: Some(vec![mock_users.clone()]), - tweets: None, - }), - meta: None, - }) - }); - - let search_api_arc: Arc = Arc::new(mock_search); - - mock_gateway.expect_search().times(1).return_const(search_api_arc); - - // --- Run Service --- - let service = TweetSynchronizerService::new( - db.clone(), - Arc::new(mock_gateway), - telegram_service, - alert_service, - config, - ); - - let result = service.sync_relevant_tweets().await; - - // --- Assertions --- - assert!(result.is_ok()); - - // 1. Check DB for Authors - let author = db.tweet_authors.find_by_id(&dummy_author.id).await.unwrap(); - assert!(author.is_some()); - assert_eq!(author.unwrap().username, dummy_author.username); - - // 2. Check DB for Tweets - let tweet = db.relevant_tweets.find_by_id("t1").await.unwrap(); - assert!(tweet.is_some()); - assert_eq!(tweet.unwrap().impression_count, 100); - } - - #[tokio::test] - async fn test_sync_sends_telegram_when_raid_active() { - let (db, mock_tg, telegram_service, alert_service, config) = setup_deps().await; - - // --- Setup DB with existing watched tweet authors --- - let dummy_author = crate::models::tweet_author::NewAuthorPayload { - id: "old_u".to_string(), - name: "Old".to_string(), - username: "old".to_string(), - followers_count: 0, - following_count: 0, - tweet_count: 0, - listed_count: 0, - like_count: 0, - media_count: 0, - is_ignored: Some(false), - }; - - db.tweet_authors.upsert(&dummy_author).await.unwrap(); - - // --- Setup Active Raid --- - db.raid_quests - .create(&CreateRaidQuest { - name: "Test Raid".to_string(), - }) - .await - .unwrap(); - - // --- Setup Telegram Mock --- - // Expect a POST to /sendMessage - Mock::given(method("POST")) - .and(path("/bot123456/sendMessage")) - .respond_with(ResponseTemplate::new(200)) - .expect(1) // We expect 1 message for 1 tweet - .mount(&mock_tg) - .await; - - // --- Setup Twitter Mock --- - let mut mock_gateway = MockTwitterGateway::new(); - let mut mock_search = MockSearchApi::new(); - - mock_search.expect_recent().returning(move |_| { - Ok(TwitterApiResponse { - data: Some(vec![create_mock_tweet("t2", &dummy_author.id)]), - includes: Some(Includes { - users: Some(vec![create_mock_user(&dummy_author.id, &dummy_author.username)]), - tweets: None, - }), - meta: None, - }) - }); - - let search_api_arc: Arc = Arc::new(mock_search); - - mock_gateway.expect_search().times(1).return_const(search_api_arc); - - // --- Run Service --- - let service = TweetSynchronizerService::new( - db.clone(), - Arc::new(mock_gateway), - telegram_service, - alert_service, - config, - ); - - // This will spawn the background task for telegram, so we need to wait slightly - // or ensure the service function awaits the spawn (it doesn't in your code). - // *Correction*: Your code uses `tokio::spawn` inside `process_sending_raid_targets`. - // We need to verify the side effect (wiremock receiving request). - service.sync_relevant_tweets().await.unwrap(); - - // Wait a bit for the spawned tokio task to fire the http request - tokio::time::sleep(tokio::time::Duration::from_millis(100)).await; - } - - /// API may repeat the `since_id` tweet; it must not be forwarded to Telegram for raids. - #[tokio::test] - async fn test_sync_excludes_last_id_from_telegram_when_raid_active() { - let (db, mock_tg, telegram_service, alert_service, config) = setup_deps().await; - - let dummy_author = crate::models::tweet_author::NewAuthorPayload { - id: "old_u".to_string(), - name: "Old".to_string(), - username: "old".to_string(), - followers_count: 0, - following_count: 0, - tweet_count: 0, - listed_count: 0, - like_count: 0, - media_count: 0, - is_ignored: Some(false), - }; - - db.tweet_authors.upsert(&dummy_author).await.unwrap(); - - let last_tweet_id = "since_id_dup"; - db.relevant_tweets - .upsert_many(&vec![crate::models::relevant_tweet::NewTweetPayload { - id: last_tweet_id.to_string(), - author_id: dummy_author.id.clone(), - text: "Already seen".to_string(), - impression_count: 0, - reply_count: 0, - retweet_count: 0, - like_count: 0, - created_at: chrono::Utc::now(), - }]) - .await - .unwrap(); - - db.raid_quests - .create(&CreateRaidQuest { - name: "Test Raid".to_string(), - }) - .await - .unwrap(); - - Mock::given(method("POST")) - .and(path("/bot123456/sendMessage")) - .respond_with(ResponseTemplate::new(200)) - .expect(0) - .mount(&mock_tg) - .await; - - let mut mock_gateway = MockTwitterGateway::new(); - let mut mock_search = MockSearchApi::new(); - - let author = dummy_author.clone(); - mock_search.expect_recent().returning(move |_| { - Ok(TwitterApiResponse { - data: Some(vec![create_mock_tweet(last_tweet_id, &author.id)]), - includes: Some(Includes { - users: Some(vec![create_mock_user(&author.id, &author.username)]), - tweets: None, - }), - meta: None, - }) - }); - - let search_api_arc: Arc = Arc::new(mock_search); - - mock_gateway.expect_search().times(1).return_const(search_api_arc); - - let service = TweetSynchronizerService::new( - db.clone(), - Arc::new(mock_gateway), - telegram_service, - alert_service, - config, - ); - - service.sync_relevant_tweets().await.unwrap(); - - tokio::time::sleep(tokio::time::Duration::from_millis(100)).await; - } - - #[tokio::test] - async fn test_pagination_logic_uses_last_id() { - let (db, _mock_tg, telegram_service, alert_service, config) = setup_deps().await; - - // --- Setup DB with existing tweet --- - // We insert a tweet directly to simulate "last state" - // Note: We need a valid author first due to FK - let dummy_author = crate::models::tweet_author::NewAuthorPayload { - id: "old_u".to_string(), - name: "Old".to_string(), - username: "old".to_string(), - followers_count: 0, - following_count: 0, - tweet_count: 0, - listed_count: 0, - like_count: 0, - media_count: 0, - is_ignored: Some(false), - }; - - db.tweet_authors.upsert(&dummy_author).await.unwrap(); - - db.relevant_tweets - .upsert_many(&vec![crate::models::relevant_tweet::NewTweetPayload { - id: "12345".to_string(), // This is the ID we expect in 'since_id' - author_id: "old_u".to_string(), - text: "Old tweet".to_string(), - impression_count: 0, - reply_count: 0, - retweet_count: 0, - like_count: 0, - created_at: chrono::Utc::now(), - }]) - .await - .unwrap(); - - // --- Setup Twitter Mock --- - let mut mock_gateway = MockTwitterGateway::new(); - let mut mock_search = MockSearchApi::new(); - - // Verify the params passed to search contain the correct `since_id` - mock_search - .expect_recent() - .withf(|params| { - // Assert that since_id matches the one in DB - params.since_id == Some("12345".to_string()) - }) - .returning(|_| { - Ok(TwitterApiResponse { - data: None, - includes: None, - meta: None, - }) - }); - - let search_api_arc: Arc = Arc::new(mock_search); - - mock_gateway.expect_search().times(1).return_const(search_api_arc); - - let service = - TweetSynchronizerService::new(db, Arc::new(mock_gateway), telegram_service, alert_service, config); - - service.sync_relevant_tweets().await.unwrap(); - } -} diff --git a/src/utils/mod.rs b/src/utils/mod.rs index eca7e60..d6d0b7e 100644 --- a/src/utils/mod.rs +++ b/src/utils/mod.rs @@ -1,6 +1,5 @@ pub mod generate_referral_code; pub mod jwt; -pub mod x_url; #[cfg(test)] pub mod test_app_state; diff --git a/src/utils/test_db.rs b/src/utils/test_db.rs index 6520d14..26020c2 100644 --- a/src/utils/test_db.rs +++ b/src/utils/test_db.rs @@ -11,7 +11,7 @@ use crate::{ }; pub async fn reset_database(pool: &PgPool) { - sqlx::query("TRUNCATE referrals, opt_ins, addresses, admins, eth_associations, x_associations, relevant_tweets, tweet_authors, raid_quests, raid_submissions, tweet_pull_usage RESTART IDENTITY CASCADE") + sqlx::query("TRUNCATE referrals, opt_ins, addresses, admins, eth_associations, x_associations, relevant_tweets, tweet_authors, raid_quests, raid_submissions RESTART IDENTITY CASCADE") .execute(pool) .await .expect("Failed to truncate tables for tests"); diff --git a/src/utils/x_url.rs b/src/utils/x_url.rs deleted file mode 100644 index 518e24d..0000000 --- a/src/utils/x_url.rs +++ /dev/null @@ -1,3 +0,0 @@ -pub fn build_x_status_url(username: &str, id: &str) -> String { - format!("https://x.com/{}/status/{}", username, id) -} From 67e1c31ef75a93effbf889ef4d7929c695c04832 Mon Sep 17 00:00:00 2001 From: Beast Date: Tue, 7 Jul 2026 13:41:56 +0800 Subject: [PATCH 3/3] fix: clippy error --- src/models/relevant_tweet.rs | 1 + src/repositories/raid_quest.rs | 45 ++++++++++++++---------------- src/repositories/relevant_tweet.rs | 9 +++--- src/repositories/tweet_author.rs | 5 ++-- 4 files changed, 30 insertions(+), 30 deletions(-) diff --git a/src/models/relevant_tweet.rs b/src/models/relevant_tweet.rs index 6ed0f9b..ac7d6b1 100644 --- a/src/models/relevant_tweet.rs +++ b/src/models/relevant_tweet.rs @@ -76,6 +76,7 @@ pub struct TweetWithAuthor { pub author_username: String, } +#[cfg(test)] #[derive(Debug, Clone, Deserialize)] pub struct NewTweetPayload { pub id: String, diff --git a/src/repositories/raid_quest.rs b/src/repositories/raid_quest.rs index bb395b0..07f36e4 100644 --- a/src/repositories/raid_quest.rs +++ b/src/repositories/raid_quest.rs @@ -14,10 +14,6 @@ pub struct RaidQuestRepository { } impl RaidQuestRepository { - fn create_select_base_query<'a>() -> QueryBuilder<'a, Postgres> { - QueryBuilder::new("SELECT * FROM raid_quests") - } - fn create_pagination_base_query<'a>( &self, query_builder: &mut QueryBuilder<'a, Postgres>, @@ -139,26 +135,6 @@ impl RaidQuestRepository { Ok(quest) } - /// Finds the single currently active raid quest. - /// Active = start_date <= NOW and (end_date IS NULL or end_date > NOW) - pub async fn find_active(&self) -> DbResult> { - let mut qb = Self::create_select_base_query(); - let now = Utc::now(); - - qb.push(" WHERE start_date <= "); - qb.push_bind(now); - - // Check that it hasn't ended yet - qb.push(" AND end_date IS NULL"); - - // Order by most recently started just to be safe - qb.push(" ORDER BY start_date DESC LIMIT 1"); - - let quest = qb.build_query_as().fetch_optional(&self.pool).await?; - - Ok(quest) - } - pub async fn finish(&self, id: i32) -> DbResult<()> { let result = sqlx::query("UPDATE raid_quests SET end_date = $1 WHERE id = $2") .bind(Utc::now()) @@ -220,6 +196,27 @@ impl RaidQuestRepository { Ok(count) } + + #[cfg(test)] + fn create_select_base_query<'a>() -> QueryBuilder<'a, Postgres> { + QueryBuilder::new("SELECT * FROM raid_quests") + } + + /// Finds the single currently active raid quest (test helper). + #[cfg(test)] + pub async fn find_active(&self) -> DbResult> { + let mut qb = Self::create_select_base_query(); + let now = Utc::now(); + + qb.push(" WHERE start_date <= "); + qb.push_bind(now); + qb.push(" AND end_date IS NULL"); + qb.push(" ORDER BY start_date DESC LIMIT 1"); + + let quest = qb.build_query_as().fetch_optional(&self.pool).await?; + + Ok(quest) + } } #[cfg(test)] diff --git a/src/repositories/relevant_tweet.rs b/src/repositories/relevant_tweet.rs index 19ab1af..dbe57ba 100644 --- a/src/repositories/relevant_tweet.rs +++ b/src/repositories/relevant_tweet.rs @@ -3,7 +3,7 @@ use sqlx::{PgPool, Postgres, QueryBuilder}; use crate::{ db_persistence::DbError, handlers::ListQueryParams, - models::relevant_tweet::{NewTweetPayload, RelevantTweet, TweetFilter, TweetSortColumn, TweetWithAuthor}, + models::relevant_tweet::{RelevantTweet, TweetFilter, TweetSortColumn, TweetWithAuthor}, repositories::{calculate_page_offset, DbResult, QueryBuilderExt}, }; @@ -136,8 +136,9 @@ impl RelevantTweetRepository { Ok(tweets) } - /// Batch Upsert - pub async fn upsert_many(&self, tweets: &Vec) -> DbResult { + /// Batch upsert used by integration tests to seed tweet data. + #[cfg(test)] + pub async fn upsert_many(&self, tweets: &[crate::models::relevant_tweet::NewTweetPayload]) -> DbResult { if tweets.is_empty() { return Ok(0); } @@ -296,7 +297,7 @@ mod tests { // Change metrics and upsert again let mut update_payload = payload.clone(); update_payload.impression_count = 999; - repo.upsert_many(&vec![update_payload]).await.unwrap(); + repo.upsert_many(&[update_payload]).await.unwrap(); let updated = repo.find_by_id(tweet_id).await.unwrap().unwrap(); assert_eq!( diff --git a/src/repositories/tweet_author.rs b/src/repositories/tweet_author.rs index 9ef5c3a..0a50b93 100644 --- a/src/repositories/tweet_author.rs +++ b/src/repositories/tweet_author.rs @@ -155,8 +155,9 @@ impl TweetAuthorRepository { Ok(id) } - /// Batch Upsert for Authors - pub async fn upsert_many(&self, authors: &Vec) -> DbResult { + /// Batch upsert used by integration tests to seed author data. + #[cfg(test)] + pub async fn upsert_many(&self, authors: &[NewAuthorPayload]) -> DbResult { if authors.is_empty() { return Ok(0); }