tuitbot_core/automation/

target_loop.rs

//! Target account monitoring loop.
//!
//! Fetches recent tweets from configured target accounts, scores them
//! with adjusted weights (preferring recency and low reply count), and
//! generates relationship-based replies. This loop operates independently
//! from keyword-based discovery to enable genuine engagement with specific
//! people.

use super::loop_helpers::{
    ConsecutiveErrorTracker, LoopError, LoopTweet, PostSender, ReplyGenerator, SafetyChecker,
};
use super::schedule::{schedule_gate, ActiveSchedule};
use super::scheduler::LoopScheduler;
use std::sync::Arc;
use std::time::Duration;
use tokio_util::sync::CancellationToken;

// ============================================================================
// Port traits specific to target loop
// ============================================================================

/// Fetches tweets from a specific user by user ID.
#[async_trait::async_trait]
pub trait TargetTweetFetcher: Send + Sync {
    /// Fetch recent tweets from the given user.
    async fn fetch_user_tweets(&self, user_id: &str) -> Result<Vec<LoopTweet>, LoopError>;
}

/// Looks up a user by username.
#[async_trait::async_trait]
pub trait TargetUserManager: Send + Sync {
    /// Look up a user by username. Returns (user_id, username).
    async fn lookup_user(&self, username: &str) -> Result<(String, String), LoopError>;
}

/// Storage operations for target account state.
#[allow(clippy::too_many_arguments)]
#[async_trait::async_trait]
pub trait TargetStorage: Send + Sync {
    /// Upsert a target account record.
    async fn upsert_target_account(
        &self,
        account_id: &str,
        username: &str,
    ) -> Result<(), LoopError>;

    /// Check if a target tweet already exists.
    async fn target_tweet_exists(&self, tweet_id: &str) -> Result<bool, LoopError>;

    /// Store a discovered target tweet.
    async fn store_target_tweet(
        &self,
        tweet_id: &str,
        account_id: &str,
        content: &str,
        created_at: &str,
        reply_count: i64,
        like_count: i64,
        relevance_score: f64,
    ) -> Result<(), LoopError>;

    /// Mark a target tweet as replied to.
    async fn mark_target_tweet_replied(&self, tweet_id: &str) -> Result<(), LoopError>;

    /// Record a reply to a target account (increments counter).
    async fn record_target_reply(&self, account_id: &str) -> Result<(), LoopError>;

    /// Get count of target replies sent today.
    async fn count_target_replies_today(&self) -> Result<i64, LoopError>;

    /// Log an action.
    async fn log_action(
        &self,
        action_type: &str,
        status: &str,
        message: &str,
    ) -> Result<(), LoopError>;
}

// ============================================================================
// Target loop config
// ============================================================================

/// Configuration for the target monitoring loop.
#[derive(Debug, Clone)]
pub struct TargetLoopConfig {
    /// Target account usernames (without @).
    pub accounts: Vec<String>,
    /// Maximum target replies per day.
    pub max_target_replies_per_day: u32,
    /// Whether this is a dry run.
    pub dry_run: bool,
}

// ============================================================================
// Target loop result
// ============================================================================

/// Result of processing a single target tweet.
#[derive(Debug)]
pub enum TargetResult {
    /// Reply was sent (or would be in dry-run).
    Replied {
        tweet_id: String,
        account: String,
        reply_text: String,
    },
    /// Tweet was skipped.
    Skipped { tweet_id: String, reason: String },
    /// Processing failed.
    Failed { tweet_id: String, error: String },
}

// ============================================================================
// Target loop
// ============================================================================

/// Monitors target accounts and generates relationship-based replies.
pub struct TargetLoop {
    fetcher: Arc<dyn TargetTweetFetcher>,
    user_mgr: Arc<dyn TargetUserManager>,
    generator: Arc<dyn ReplyGenerator>,
    safety: Arc<dyn SafetyChecker>,
    storage: Arc<dyn TargetStorage>,
    poster: Arc<dyn PostSender>,
    config: TargetLoopConfig,
}

impl TargetLoop {
    /// Create a new target monitoring loop.
    #[allow(clippy::too_many_arguments)]
    pub fn new(
        fetcher: Arc<dyn TargetTweetFetcher>,
        user_mgr: Arc<dyn TargetUserManager>,
        generator: Arc<dyn ReplyGenerator>,
        safety: Arc<dyn SafetyChecker>,
        storage: Arc<dyn TargetStorage>,
        poster: Arc<dyn PostSender>,
        config: TargetLoopConfig,
    ) -> Self {
        Self {
            fetcher,
            user_mgr,
            generator,
            safety,
            storage,
            poster,
            config,
        }
    }

    /// Run the continuous target monitoring loop until cancellation.
    pub async fn run(
        &self,
        cancel: CancellationToken,
        scheduler: LoopScheduler,
        schedule: Option<Arc<ActiveSchedule>>,
    ) {
        tracing::info!(
            dry_run = self.config.dry_run,
            accounts = self.config.accounts.len(),
            max_replies = self.config.max_target_replies_per_day,
            "Target monitoring loop started"
        );

        if self.config.accounts.is_empty() {
            tracing::info!("No target accounts configured, target loop has nothing to do");
            cancel.cancelled().await;
            return;
        }

        let mut error_tracker = ConsecutiveErrorTracker::new(10, Duration::from_secs(300));

        loop {
            if cancel.is_cancelled() {
                break;
            }

            if !schedule_gate(&schedule, &cancel).await {
                break;
            }

            match self.run_iteration().await {
                Ok(results) => {
                    error_tracker.record_success();
                    let replied = results
                        .iter()
                        .filter(|r| matches!(r, TargetResult::Replied { .. }))
                        .count();
                    let skipped = results
                        .iter()
                        .filter(|r| matches!(r, TargetResult::Skipped { .. }))
                        .count();
                    if !results.is_empty() {
                        tracing::info!(
                            total = results.len(),
                            replied = replied,
                            skipped = skipped,
                            "Target iteration complete"
                        );
                    }
                }
                Err(e) => {
                    let should_pause = error_tracker.record_error();
                    tracing::warn!(
                        error = %e,
                        consecutive_errors = error_tracker.count(),
                        "Target iteration failed"
                    );

                    if should_pause {
                        tracing::warn!(
                            pause_secs = error_tracker.pause_duration().as_secs(),
                            "Pausing target loop due to consecutive errors"
                        );
                        tokio::select! {
                            _ = cancel.cancelled() => break,
                            _ = tokio::time::sleep(error_tracker.pause_duration()) => {},
                        }
                        error_tracker.reset();
                        continue;
                    }
                }
            }

            tokio::select! {
                _ = cancel.cancelled() => break,
                _ = scheduler.tick() => {},
            }
        }

        tracing::info!("Target monitoring loop stopped");
    }

    /// Run a single iteration across all target accounts.
    pub async fn run_iteration(&self) -> Result<Vec<TargetResult>, LoopError> {
        let mut all_results = Vec::new();

        // Check daily limit
        let replies_today = self.storage.count_target_replies_today().await?;
        if replies_today >= self.config.max_target_replies_per_day as i64 {
            tracing::debug!(
                replies_today = replies_today,
                limit = self.config.max_target_replies_per_day,
                "Target reply daily limit reached"
            );
            return Ok(all_results);
        }

        let mut remaining_replies =
            (self.config.max_target_replies_per_day as i64 - replies_today) as usize;

        for username in &self.config.accounts {
            if remaining_replies == 0 {
                break;
            }

            match self.process_account(username, remaining_replies).await {
                Ok(results) => {
                    let replied_count = results
                        .iter()
                        .filter(|r| matches!(r, TargetResult::Replied { .. }))
                        .count();
                    remaining_replies = remaining_replies.saturating_sub(replied_count);
                    all_results.extend(results);
                }
                Err(e) => {
                    // AuthExpired is global — stop immediately instead of
                    // failing N times with the same 401.
                    if matches!(e, LoopError::AuthExpired) {
                        tracing::error!(
                            username = %username,
                            "X API authentication expired, re-authenticate with `tuitbot init`"
                        );
                        return Err(e);
                    }

                    tracing::warn!(
                        username = %username,
                        error = %e,
                        "Failed to process target account"
                    );
                }
            }
        }

        Ok(all_results)
    }

    /// Process a single target account: resolve, fetch tweets, reply.
    async fn process_account(
        &self,
        username: &str,
        max_replies: usize,
    ) -> Result<Vec<TargetResult>, LoopError> {
        // Look up user
        let (user_id, resolved_username) = self.user_mgr.lookup_user(username).await?;

        // Upsert target account record
        self.storage
            .upsert_target_account(&user_id, &resolved_username)
            .await?;

        // Fetch recent tweets
        let tweets = self.fetcher.fetch_user_tweets(&user_id).await?;
        tracing::info!(
            username = %resolved_username,
            count = tweets.len(),
            "Monitoring @{}, found {} new tweets",
            resolved_username,
            tweets.len(),
        );

        let mut results = Vec::new();

        for tweet in tweets.iter().take(max_replies) {
            let result = self
                .process_target_tweet(tweet, &user_id, &resolved_username)
                .await;
            if matches!(result, TargetResult::Replied { .. }) {
                results.push(result);
                // Only reply to one tweet per account per iteration
                break;
            }
            results.push(result);
        }

        Ok(results)
    }

    /// Process a single target tweet: dedup, safety check, generate reply, post.
    async fn process_target_tweet(
        &self,
        tweet: &LoopTweet,
        account_id: &str,
        username: &str,
    ) -> TargetResult {
        // Check if already seen
        match self.storage.target_tweet_exists(&tweet.id).await {
            Ok(true) => {
                return TargetResult::Skipped {
                    tweet_id: tweet.id.clone(),
                    reason: "already discovered".to_string(),
                };
            }
            Ok(false) => {}
            Err(e) => {
                tracing::warn!(tweet_id = %tweet.id, error = %e, "Failed to check target tweet");
            }
        }

        // Store the discovered tweet
        let _ = self
            .storage
            .store_target_tweet(
                &tweet.id,
                account_id,
                &tweet.text,
                &tweet.created_at,
                tweet.replies as i64,
                tweet.likes as i64,
                0.0,
            )
            .await;

        // Safety checks
        if self.safety.has_replied_to(&tweet.id).await {
            return TargetResult::Skipped {
                tweet_id: tweet.id.clone(),
                reason: "already replied".to_string(),
            };
        }

        if !self.safety.can_reply().await {
            return TargetResult::Skipped {
                tweet_id: tweet.id.clone(),
                reason: "rate limited".to_string(),
            };
        }

        // Generate reply (no product mention for target accounts — be genuine)
        let reply_text = match self
            .generator
            .generate_reply(&tweet.text, username, false)
            .await
        {
            Ok(text) => text,
            Err(e) => {
                return TargetResult::Failed {
                    tweet_id: tweet.id.clone(),
                    error: e.to_string(),
                };
            }
        };

        tracing::info!(
            username = %username,
            "Replied to target @{}",
            username,
        );

        if self.config.dry_run {
            tracing::info!(
                "DRY RUN: Target @{} tweet {} -- Would reply: \"{}\"",
                username,
                tweet.id,
                reply_text
            );

            let _ = self
                .storage
                .log_action(
                    "target_reply",
                    "dry_run",
                    &format!("Reply to @{username}: {}", truncate(&reply_text, 50)),
                )
                .await;
        } else {
            if let Err(e) = self.poster.send_reply(&tweet.id, &reply_text).await {
                return TargetResult::Failed {
                    tweet_id: tweet.id.clone(),
                    error: e.to_string(),
                };
            }

            if let Err(e) = self.safety.record_reply(&tweet.id, &reply_text).await {
                tracing::warn!(tweet_id = %tweet.id, error = %e, "Failed to record reply");
            }

            // Mark tweet as replied and update account stats
            let _ = self.storage.mark_target_tweet_replied(&tweet.id).await;
            let _ = self.storage.record_target_reply(account_id).await;

            let _ = self
                .storage
                .log_action(
                    "target_reply",
                    "success",
                    &format!("Replied to @{username}: {}", truncate(&reply_text, 50)),
                )
                .await;
        }

        TargetResult::Replied {
            tweet_id: tweet.id.clone(),
            account: username.to_string(),
            reply_text,
        }
    }
}

/// Truncate a string for display.
fn truncate(s: &str, max_len: usize) -> String {
    if s.len() <= max_len {
        s.to_string()
    } else {
        format!("{}...", &s[..max_len])
    }
}

#[cfg(test)]
mod tests {
    use super::*;
    use std::sync::atomic::{AtomicU32, Ordering};
    use std::sync::Mutex;

    // --- Mock implementations ---

    struct MockFetcher {
        tweets: Vec<LoopTweet>,
    }

    #[async_trait::async_trait]
    impl TargetTweetFetcher for MockFetcher {
        async fn fetch_user_tweets(&self, _user_id: &str) -> Result<Vec<LoopTweet>, LoopError> {
            Ok(self.tweets.clone())
        }
    }

    struct MockUserManager {
        users: Vec<(String, String, String)>, // (username, user_id, resolved_username)
    }

    #[async_trait::async_trait]
    impl TargetUserManager for MockUserManager {
        async fn lookup_user(&self, username: &str) -> Result<(String, String), LoopError> {
            for (uname, uid, resolved) in &self.users {
                if uname == username {
                    return Ok((uid.clone(), resolved.clone()));
                }
            }
            Err(LoopError::Other(format!("user not found: {username}")))
        }
    }

    struct MockGenerator {
        reply: String,
    }

    #[async_trait::async_trait]
    impl ReplyGenerator for MockGenerator {
        async fn generate_reply(
            &self,
            _tweet_text: &str,
            _author: &str,
            _mention_product: bool,
        ) -> Result<String, LoopError> {
            Ok(self.reply.clone())
        }
    }

    struct MockSafety {
        can_reply: bool,
        replied_ids: Mutex<Vec<String>>,
    }

    impl MockSafety {
        fn new(can_reply: bool) -> Self {
            Self {
                can_reply,
                replied_ids: Mutex::new(Vec::new()),
            }
        }
    }

    #[async_trait::async_trait]
    impl SafetyChecker for MockSafety {
        async fn can_reply(&self) -> bool {
            self.can_reply
        }
        async fn has_replied_to(&self, tweet_id: &str) -> bool {
            self.replied_ids
                .lock()
                .expect("lock")
                .contains(&tweet_id.to_string())
        }
        async fn record_reply(&self, tweet_id: &str, _content: &str) -> Result<(), LoopError> {
            self.replied_ids
                .lock()
                .expect("lock")
                .push(tweet_id.to_string());
            Ok(())
        }
    }

    struct MockTargetStorage {
        existing_tweets: Mutex<Vec<String>>,
        replies_today: Mutex<i64>,
    }

    impl MockTargetStorage {
        fn new() -> Self {
            Self {
                existing_tweets: Mutex::new(Vec::new()),
                replies_today: Mutex::new(0),
            }
        }
    }

    #[async_trait::async_trait]
    impl TargetStorage for MockTargetStorage {
        async fn upsert_target_account(
            &self,
            _account_id: &str,
            _username: &str,
        ) -> Result<(), LoopError> {
            Ok(())
        }
        async fn target_tweet_exists(&self, tweet_id: &str) -> Result<bool, LoopError> {
            Ok(self
                .existing_tweets
                .lock()
                .expect("lock")
                .contains(&tweet_id.to_string()))
        }
        async fn store_target_tweet(
            &self,
            _tweet_id: &str,
            _account_id: &str,
            _content: &str,
            _created_at: &str,
            _reply_count: i64,
            _like_count: i64,
            _relevance_score: f64,
        ) -> Result<(), LoopError> {
            Ok(())
        }
        async fn mark_target_tweet_replied(&self, _tweet_id: &str) -> Result<(), LoopError> {
            Ok(())
        }
        async fn record_target_reply(&self, _account_id: &str) -> Result<(), LoopError> {
            *self.replies_today.lock().expect("lock") += 1;
            Ok(())
        }
        async fn count_target_replies_today(&self) -> Result<i64, LoopError> {
            Ok(*self.replies_today.lock().expect("lock"))
        }
        async fn log_action(
            &self,
            _action_type: &str,
            _status: &str,
            _message: &str,
        ) -> Result<(), LoopError> {
            Ok(())
        }
    }

    struct MockPoster {
        sent: Mutex<Vec<(String, String)>>,
    }

    impl MockPoster {
        fn new() -> Self {
            Self {
                sent: Mutex::new(Vec::new()),
            }
        }
        fn sent_count(&self) -> usize {
            self.sent.lock().expect("lock").len()
        }
    }

    #[async_trait::async_trait]
    impl PostSender for MockPoster {
        async fn send_reply(&self, tweet_id: &str, content: &str) -> Result<(), LoopError> {
            self.sent
                .lock()
                .expect("lock")
                .push((tweet_id.to_string(), content.to_string()));
            Ok(())
        }
    }

    fn test_tweet(id: &str, author: &str) -> LoopTweet {
        LoopTweet {
            id: id.to_string(),
            text: format!("Interesting thoughts on tech from @{author}"),
            author_id: format!("uid_{author}"),
            author_username: author.to_string(),
            author_followers: 5000,
            created_at: "2026-01-01T00:00:00Z".to_string(),
            likes: 10,
            retweets: 2,
            replies: 1,
        }
    }

    fn default_config() -> TargetLoopConfig {
        TargetLoopConfig {
            accounts: vec!["alice".to_string()],
            max_target_replies_per_day: 3,
            dry_run: false,
        }
    }

    fn build_loop(
        tweets: Vec<LoopTweet>,
        config: TargetLoopConfig,
        storage: Arc<MockTargetStorage>,
    ) -> (TargetLoop, Arc<MockPoster>) {
        let poster = Arc::new(MockPoster::new());
        let user_mgr = Arc::new(MockUserManager {
            users: vec![(
                "alice".to_string(),
                "uid_alice".to_string(),
                "alice".to_string(),
            )],
        });
        let target_loop = TargetLoop::new(
            Arc::new(MockFetcher { tweets }),
            user_mgr,
            Arc::new(MockGenerator {
                reply: "Great point!".to_string(),
            }),
            Arc::new(MockSafety::new(true)),
            storage,
            poster.clone(),
            config,
        );
        (target_loop, poster)
    }

    // --- Tests ---

    #[tokio::test]
    async fn empty_accounts_does_nothing() {
        let storage = Arc::new(MockTargetStorage::new());
        let mut config = default_config();
        config.accounts = Vec::new();
        let (target_loop, poster) = build_loop(Vec::new(), config, storage);

        let results = target_loop.run_iteration().await.expect("iteration");
        assert!(results.is_empty());
        assert_eq!(poster.sent_count(), 0);
    }

    #[tokio::test]
    async fn replies_to_target_tweet() {
        let tweets = vec![test_tweet("tw1", "alice")];
        let storage = Arc::new(MockTargetStorage::new());
        let (target_loop, poster) = build_loop(tweets, default_config(), storage);

        let results = target_loop.run_iteration().await.expect("iteration");
        assert_eq!(results.len(), 1);
        assert!(matches!(results[0], TargetResult::Replied { .. }));
        assert_eq!(poster.sent_count(), 1);
    }

    #[tokio::test]
    async fn skips_existing_target_tweet() {
        let tweets = vec![test_tweet("tw1", "alice")];
        let storage = Arc::new(MockTargetStorage::new());
        storage
            .existing_tweets
            .lock()
            .expect("lock")
            .push("tw1".to_string());
        let (target_loop, poster) = build_loop(tweets, default_config(), storage);

        let results = target_loop.run_iteration().await.expect("iteration");
        assert_eq!(results.len(), 1);
        assert!(matches!(results[0], TargetResult::Skipped { .. }));
        assert_eq!(poster.sent_count(), 0);
    }

    #[tokio::test]
    async fn respects_daily_limit() {
        let tweets = vec![test_tweet("tw1", "alice")];
        let storage = Arc::new(MockTargetStorage::new());
        *storage.replies_today.lock().expect("lock") = 3;
        let (target_loop, poster) = build_loop(tweets, default_config(), storage);

        let results = target_loop.run_iteration().await.expect("iteration");
        assert!(results.is_empty());
        assert_eq!(poster.sent_count(), 0);
    }

    #[tokio::test]
    async fn dry_run_does_not_post() {
        let tweets = vec![test_tweet("tw1", "alice")];
        let storage = Arc::new(MockTargetStorage::new());
        let mut config = default_config();
        config.dry_run = true;
        let (target_loop, poster) = build_loop(tweets, config, storage);

        let results = target_loop.run_iteration().await.expect("iteration");
        assert_eq!(results.len(), 1);
        assert!(matches!(results[0], TargetResult::Replied { .. }));
        assert_eq!(poster.sent_count(), 0);
    }

    #[test]
    fn truncate_short_string() {
        assert_eq!(truncate("hello", 10), "hello");
    }

    #[test]
    fn truncate_long_string() {
        assert_eq!(truncate("hello world", 5), "hello...");
    }

    // --- AuthExpired-aware mock ---

    struct MockAuthExpiredUserManager {
        lookup_count: AtomicU32,
        error: LoopError,
    }

    impl MockAuthExpiredUserManager {
        fn auth_expired() -> Self {
            Self {
                lookup_count: AtomicU32::new(0),
                error: LoopError::AuthExpired,
            }
        }
    }

    #[async_trait::async_trait]
    impl TargetUserManager for MockAuthExpiredUserManager {
        async fn lookup_user(&self, _username: &str) -> Result<(String, String), LoopError> {
            self.lookup_count.fetch_add(1, Ordering::SeqCst);
            Err(match &self.error {
                LoopError::AuthExpired => LoopError::AuthExpired,
                LoopError::Other(msg) => LoopError::Other(msg.clone()),
                _ => unreachable!(),
            })
        }
    }

    /// A user manager where the first lookup fails and the second succeeds.
    struct MockPartialFailUserManager {
        call_count: AtomicU32,
    }

    #[async_trait::async_trait]
    impl TargetUserManager for MockPartialFailUserManager {
        async fn lookup_user(&self, username: &str) -> Result<(String, String), LoopError> {
            let n = self.call_count.fetch_add(1, Ordering::SeqCst);
            if n == 0 {
                Err(LoopError::Other("transient failure".to_string()))
            } else {
                Ok((format!("uid_{username}"), username.to_string()))
            }
        }
    }

    #[tokio::test]
    async fn auth_expired_stops_iteration() {
        let user_mgr = Arc::new(MockAuthExpiredUserManager::auth_expired());
        let storage = Arc::new(MockTargetStorage::new());
        let poster = Arc::new(MockPoster::new());

        let mut config = default_config();
        config.accounts = vec![
            "alice".to_string(),
            "bob".to_string(),
            "charlie".to_string(),
        ];

        let target_loop = TargetLoop::new(
            Arc::new(MockFetcher { tweets: vec![] }),
            user_mgr.clone(),
            Arc::new(MockGenerator {
                reply: "Great!".to_string(),
            }),
            Arc::new(MockSafety::new(true)),
            storage,
            poster,
            config,
        );

        let result = target_loop.run_iteration().await;
        assert!(result.is_err());
        assert!(matches!(result.unwrap_err(), LoopError::AuthExpired));
        // Only one lookup should have been attempted — the loop exits early.
        assert_eq!(user_mgr.lookup_count.load(Ordering::SeqCst), 1);
    }

    #[tokio::test]
    async fn non_auth_error_continues_iteration() {
        let user_mgr = Arc::new(MockPartialFailUserManager {
            call_count: AtomicU32::new(0),
        });
        let storage = Arc::new(MockTargetStorage::new());
        let poster = Arc::new(MockPoster::new());

        let mut config = default_config();
        config.accounts = vec!["alice".to_string(), "bob".to_string()];

        let target_loop = TargetLoop::new(
            Arc::new(MockFetcher {
                tweets: vec![test_tweet("tw1", "bob")],
            }),
            user_mgr.clone(),
            Arc::new(MockGenerator {
                reply: "Nice!".to_string(),
            }),
            Arc::new(MockSafety::new(true)),
            storage,
            poster.clone(),
            config,
        );

        let results = target_loop.run_iteration().await.expect("should succeed");
        // First account fails with Other, second succeeds — both should be attempted.
        assert_eq!(user_mgr.call_count.load(Ordering::SeqCst), 2);
        // Second account produces a reply.
        assert_eq!(results.len(), 1);
        assert!(matches!(results[0], TargetResult::Replied { .. }));
        assert_eq!(poster.sent_count(), 1);
    }
}