tuitbot_core/storage/

analytics.rs

//! CRUD operations for analytics tables.
//!
//! Manages follower snapshots, reply/tweet performance metrics,
//! and content score running averages.

use super::accounts::DEFAULT_ACCOUNT_ID;
use super::DbPool;
use crate::error::StorageError;
use chrono::{NaiveDate, Utc};

// ============================================================================
// Follower snapshots
// ============================================================================

/// A daily follower snapshot.
#[derive(Debug, Clone, serde::Serialize)]
pub struct FollowerSnapshot {
    pub snapshot_date: String,
    pub follower_count: i64,
    pub following_count: i64,
    pub tweet_count: i64,
}

/// Upsert a follower snapshot for today for a specific account.
pub async fn upsert_follower_snapshot_for(
    pool: &DbPool,
    account_id: &str,
    follower_count: i64,
    following_count: i64,
    tweet_count: i64,
) -> Result<(), StorageError> {
    sqlx::query(
        "INSERT INTO follower_snapshots (account_id, snapshot_date, follower_count, following_count, tweet_count) \
         VALUES (?, date('now'), ?, ?, ?) \
         ON CONFLICT(snapshot_date) DO UPDATE SET \
         account_id = excluded.account_id, \
         follower_count = excluded.follower_count, \
         following_count = excluded.following_count, \
         tweet_count = excluded.tweet_count",
    )
    .bind(account_id)
    .bind(follower_count)
    .bind(following_count)
    .bind(tweet_count)
    .execute(pool)
    .await
    .map_err(|e| StorageError::Query { source: e })?;
    Ok(())
}

/// Upsert a follower snapshot for today.
pub async fn upsert_follower_snapshot(
    pool: &DbPool,
    follower_count: i64,
    following_count: i64,
    tweet_count: i64,
) -> Result<(), StorageError> {
    upsert_follower_snapshot_for(
        pool,
        DEFAULT_ACCOUNT_ID,
        follower_count,
        following_count,
        tweet_count,
    )
    .await
}

/// Get the most recent N follower snapshots for a specific account, newest first.
pub async fn get_follower_snapshots_for(
    pool: &DbPool,
    account_id: &str,
    limit: u32,
) -> Result<Vec<FollowerSnapshot>, StorageError> {
    let rows: Vec<(String, i64, i64, i64)> = sqlx::query_as(
        "SELECT snapshot_date, follower_count, following_count, tweet_count \
         FROM follower_snapshots WHERE account_id = ? ORDER BY snapshot_date DESC LIMIT ?",
    )
    .bind(account_id)
    .bind(limit)
    .fetch_all(pool)
    .await
    .map_err(|e| StorageError::Query { source: e })?;

    Ok(rows
        .into_iter()
        .map(|r| FollowerSnapshot {
            snapshot_date: r.0,
            follower_count: r.1,
            following_count: r.2,
            tweet_count: r.3,
        })
        .collect())
}

/// Get the most recent N follower snapshots, newest first.
pub async fn get_follower_snapshots(
    pool: &DbPool,
    limit: u32,
) -> Result<Vec<FollowerSnapshot>, StorageError> {
    get_follower_snapshots_for(pool, DEFAULT_ACCOUNT_ID, limit).await
}

// ============================================================================
// Reply performance
// ============================================================================

/// Store or update reply performance metrics for a specific account.
pub async fn upsert_reply_performance_for(
    pool: &DbPool,
    account_id: &str,
    reply_id: &str,
    likes: i64,
    replies: i64,
    impressions: i64,
    score: f64,
) -> Result<(), StorageError> {
    sqlx::query(
        "INSERT INTO reply_performance (account_id, reply_id, likes_received, replies_received, impressions, performance_score) \
         VALUES (?, ?, ?, ?, ?, ?) \
         ON CONFLICT(reply_id) DO UPDATE SET \
         likes_received = excluded.likes_received, \
         replies_received = excluded.replies_received, \
         impressions = excluded.impressions, \
         performance_score = excluded.performance_score, \
         measured_at = datetime('now')",
    )
    .bind(account_id)
    .bind(reply_id)
    .bind(likes)
    .bind(replies)
    .bind(impressions)
    .bind(score)
    .execute(pool)
    .await
    .map_err(|e| StorageError::Query { source: e })?;
    Ok(())
}

/// Store or update reply performance metrics.
pub async fn upsert_reply_performance(
    pool: &DbPool,
    reply_id: &str,
    likes: i64,
    replies: i64,
    impressions: i64,
    score: f64,
) -> Result<(), StorageError> {
    upsert_reply_performance_for(
        pool,
        DEFAULT_ACCOUNT_ID,
        reply_id,
        likes,
        replies,
        impressions,
        score,
    )
    .await
}

// ============================================================================
// Tweet performance
// ============================================================================

/// Store or update tweet performance metrics for a specific account.
#[allow(clippy::too_many_arguments)]
pub async fn upsert_tweet_performance_for(
    pool: &DbPool,
    account_id: &str,
    tweet_id: &str,
    likes: i64,
    retweets: i64,
    replies: i64,
    impressions: i64,
    score: f64,
) -> Result<(), StorageError> {
    sqlx::query(
        "INSERT INTO tweet_performance (account_id, tweet_id, likes_received, retweets_received, replies_received, impressions, performance_score) \
         VALUES (?, ?, ?, ?, ?, ?, ?) \
         ON CONFLICT(tweet_id) DO UPDATE SET \
         likes_received = excluded.likes_received, \
         retweets_received = excluded.retweets_received, \
         replies_received = excluded.replies_received, \
         impressions = excluded.impressions, \
         performance_score = excluded.performance_score, \
         measured_at = datetime('now')",
    )
    .bind(account_id)
    .bind(tweet_id)
    .bind(likes)
    .bind(retweets)
    .bind(replies)
    .bind(impressions)
    .bind(score)
    .execute(pool)
    .await
    .map_err(|e| StorageError::Query { source: e })?;
    Ok(())
}

/// Store or update tweet performance metrics.
pub async fn upsert_tweet_performance(
    pool: &DbPool,
    tweet_id: &str,
    likes: i64,
    retweets: i64,
    replies: i64,
    impressions: i64,
    score: f64,
) -> Result<(), StorageError> {
    upsert_tweet_performance_for(
        pool,
        DEFAULT_ACCOUNT_ID,
        tweet_id,
        likes,
        retweets,
        replies,
        impressions,
        score,
    )
    .await
}

// ============================================================================
// Content scores
// ============================================================================

/// A topic/format performance score.
#[derive(Debug, Clone, serde::Serialize)]
pub struct ContentScore {
    pub topic: String,
    pub format: String,
    pub total_posts: i64,
    pub avg_performance: f64,
}

/// Update the running average for a topic/format pair for a specific account.
///
/// Uses incremental mean: new_avg = old_avg + (score - old_avg) / new_count.
pub async fn update_content_score_for(
    pool: &DbPool,
    account_id: &str,
    topic: &str,
    format: &str,
    new_score: f64,
) -> Result<(), StorageError> {
    // Insert or update with incremental average
    sqlx::query(
        "INSERT INTO content_scores (account_id, topic, format, total_posts, avg_performance) \
         VALUES (?, ?, ?, 1, ?) \
         ON CONFLICT(topic, format) DO UPDATE SET \
         account_id = excluded.account_id, \
         total_posts = content_scores.total_posts + 1, \
         avg_performance = content_scores.avg_performance + \
         (? - content_scores.avg_performance) / (content_scores.total_posts + 1)",
    )
    .bind(account_id)
    .bind(topic)
    .bind(format)
    .bind(new_score)
    .bind(new_score)
    .execute(pool)
    .await
    .map_err(|e| StorageError::Query { source: e })?;
    Ok(())
}

/// Update the running average for a topic/format pair.
///
/// Uses incremental mean: new_avg = old_avg + (score - old_avg) / new_count.
pub async fn update_content_score(
    pool: &DbPool,
    topic: &str,
    format: &str,
    new_score: f64,
) -> Result<(), StorageError> {
    update_content_score_for(pool, DEFAULT_ACCOUNT_ID, topic, format, new_score).await
}

/// Get top-performing topics for a specific account ordered by average performance descending.
pub async fn get_top_topics_for(
    pool: &DbPool,
    account_id: &str,
    limit: u32,
) -> Result<Vec<ContentScore>, StorageError> {
    let rows: Vec<(String, String, i64, f64)> = sqlx::query_as(
        "SELECT topic, format, total_posts, avg_performance \
         FROM content_scores \
         WHERE account_id = ? \
         ORDER BY avg_performance DESC \
         LIMIT ?",
    )
    .bind(account_id)
    .bind(limit)
    .fetch_all(pool)
    .await
    .map_err(|e| StorageError::Query { source: e })?;

    Ok(rows
        .into_iter()
        .map(|r| ContentScore {
            topic: r.0,
            format: r.1,
            total_posts: r.2,
            avg_performance: r.3,
        })
        .collect())
}

/// Get top-performing topics ordered by average performance descending.
pub async fn get_top_topics(pool: &DbPool, limit: u32) -> Result<Vec<ContentScore>, StorageError> {
    get_top_topics_for(pool, DEFAULT_ACCOUNT_ID, limit).await
}

/// Get average reply engagement rate for a specific account (avg performance_score across all measured replies).
pub async fn get_avg_reply_engagement_for(
    pool: &DbPool,
    account_id: &str,
) -> Result<f64, StorageError> {
    let row: (f64,) = sqlx::query_as(
        "SELECT COALESCE(AVG(performance_score), 0.0) FROM reply_performance WHERE account_id = ?",
    )
    .bind(account_id)
    .fetch_one(pool)
    .await
    .map_err(|e| StorageError::Query { source: e })?;

    Ok(row.0)
}

/// Get average reply engagement rate (avg performance_score across all measured replies).
pub async fn get_avg_reply_engagement(pool: &DbPool) -> Result<f64, StorageError> {
    get_avg_reply_engagement_for(pool, DEFAULT_ACCOUNT_ID).await
}

/// Get average tweet engagement rate for a specific account (avg performance_score across all measured tweets).
pub async fn get_avg_tweet_engagement_for(
    pool: &DbPool,
    account_id: &str,
) -> Result<f64, StorageError> {
    let row: (f64,) = sqlx::query_as(
        "SELECT COALESCE(AVG(performance_score), 0.0) FROM tweet_performance WHERE account_id = ?",
    )
    .bind(account_id)
    .fetch_one(pool)
    .await
    .map_err(|e| StorageError::Query { source: e })?;

    Ok(row.0)
}

/// Get average tweet engagement rate (avg performance_score across all measured tweets).
pub async fn get_avg_tweet_engagement(pool: &DbPool) -> Result<f64, StorageError> {
    get_avg_tweet_engagement_for(pool, DEFAULT_ACCOUNT_ID).await
}

/// Get total count of measured replies and tweets for a specific account.
pub async fn get_performance_counts_for(
    pool: &DbPool,
    account_id: &str,
) -> Result<(i64, i64), StorageError> {
    let reply_count: (i64,) =
        sqlx::query_as("SELECT COUNT(*) FROM reply_performance WHERE account_id = ?")
            .bind(account_id)
            .fetch_one(pool)
            .await
            .map_err(|e| StorageError::Query { source: e })?;

    let tweet_count: (i64,) =
        sqlx::query_as("SELECT COUNT(*) FROM tweet_performance WHERE account_id = ?")
            .bind(account_id)
            .fetch_one(pool)
            .await
            .map_err(|e| StorageError::Query { source: e })?;

    Ok((reply_count.0, tweet_count.0))
}

/// Get total count of measured replies and tweets.
pub async fn get_performance_counts(pool: &DbPool) -> Result<(i64, i64), StorageError> {
    get_performance_counts_for(pool, DEFAULT_ACCOUNT_ID).await
}

/// Compute the performance score for a piece of content.
///
/// Formula: `(likes * 3 + replies * 5 + retweets * 4) / max(impressions, 1) * 1000`
pub fn compute_performance_score(likes: i64, replies: i64, retweets: i64, impressions: i64) -> f64 {
    let numerator = (likes * 3 + replies * 5 + retweets * 4) as f64;
    let denominator = impressions.max(1) as f64;
    numerator / denominator * 1000.0
}

// ============================================================================
// Analytics summary (aggregated dashboard data)
// ============================================================================

/// Follower growth metrics.
#[derive(Debug, Clone, serde::Serialize)]
pub struct FollowerSummary {
    pub current: i64,
    pub change_7d: i64,
    pub change_30d: i64,
}

/// Today's action breakdown.
#[derive(Debug, Clone, serde::Serialize)]
pub struct ActionsSummary {
    pub replies: i64,
    pub tweets: i64,
    pub threads: i64,
}

/// Engagement overview.
#[derive(Debug, Clone, serde::Serialize)]
pub struct EngagementSummary {
    pub avg_reply_score: f64,
    pub avg_tweet_score: f64,
    pub total_replies_sent: i64,
    pub total_tweets_posted: i64,
}

/// Combined analytics summary for the dashboard.
#[derive(Debug, Clone, serde::Serialize)]
pub struct AnalyticsSummary {
    pub followers: FollowerSummary,
    pub actions_today: ActionsSummary,
    pub engagement: EngagementSummary,
    pub top_topics: Vec<ContentScore>,
}

/// Get a combined analytics summary for the dashboard for a specific account.
///
/// Aggregates follower deltas, today's action counts, and engagement stats
/// into a single struct to minimise round-trips from the frontend.
pub async fn get_analytics_summary_for(
    pool: &DbPool,
    account_id: &str,
) -> Result<AnalyticsSummary, StorageError> {
    // --- Follower data ---
    let snapshots = get_follower_snapshots_for(pool, account_id, 90).await?;
    let current = snapshots.first().map_or(0, |s| s.follower_count);

    // Find the first snapshot whose date is at least N days ago (handles gaps from
    // downtime or weekends).  Snapshots are ordered newest-first.
    let today = Utc::now().date_naive();
    let follower_at_or_before = |days: i64| -> i64 {
        snapshots
            .iter()
            .find(|s| {
                NaiveDate::parse_from_str(&s.snapshot_date, "%Y-%m-%d")
                    .map(|d| (today - d).num_days() >= days)
                    .unwrap_or(false)
            })
            .map_or(current, |s| s.follower_count)
    };

    let change_7d = if snapshots.len() >= 2 {
        current - follower_at_or_before(7)
    } else {
        0
    };
    let change_30d = if snapshots.len() >= 2 {
        current - follower_at_or_before(30)
    } else {
        0
    };

    // --- Today's actions (from action_log) ---
    let today = Utc::now().format("%Y-%m-%dT00:00:00Z").to_string();
    let counts = super::action_log::get_action_counts_since(pool, &today).await?;
    let actions_today = ActionsSummary {
        replies: *counts.get("reply").unwrap_or(&0),
        tweets: *counts.get("tweet").unwrap_or(&0),
        threads: *counts.get("thread").unwrap_or(&0),
    };

    // --- Engagement ---
    let avg_reply_score = get_avg_reply_engagement_for(pool, account_id).await?;
    let avg_tweet_score = get_avg_tweet_engagement_for(pool, account_id).await?;
    let (total_replies_sent, total_tweets_posted) =
        get_performance_counts_for(pool, account_id).await?;

    // --- Top topics ---
    let top_topics = get_top_topics_for(pool, account_id, 5).await?;

    Ok(AnalyticsSummary {
        followers: FollowerSummary {
            current,
            change_7d,
            change_30d,
        },
        actions_today,
        engagement: EngagementSummary {
            avg_reply_score,
            avg_tweet_score,
            total_replies_sent,
            total_tweets_posted,
        },
        top_topics,
    })
}

/// Get a combined analytics summary for the dashboard.
///
/// Aggregates follower deltas, today's action counts, and engagement stats
/// into a single struct to minimise round-trips from the frontend.
pub async fn get_analytics_summary(pool: &DbPool) -> Result<AnalyticsSummary, StorageError> {
    get_analytics_summary_for(pool, DEFAULT_ACCOUNT_ID).await
}

// ============================================================================
// Recent performance (joined with content for preview)
// ============================================================================

/// A recent content item with performance metrics.
#[derive(Debug, Clone, serde::Serialize)]
pub struct PerformanceItem {
    /// "reply", "tweet", or "thread"
    pub content_type: String,
    /// Truncated content preview
    pub content_preview: String,
    /// Likes received
    pub likes: i64,
    /// Replies received
    pub replies_received: i64,
    /// Retweets (0 for replies)
    pub retweets: i64,
    /// Impressions
    pub impressions: i64,
    /// Computed performance score
    pub performance_score: f64,
    /// When the content was posted (ISO-8601)
    pub posted_at: String,
}

/// Row type returned by the recent-performance UNION query.
type PerformanceRow = (String, String, i64, i64, i64, i64, f64, String);

/// Get recent content performance items for a specific account, newest first.
///
/// Unions reply and tweet performance joined with their content tables
/// so the dashboard can show a content preview alongside metrics.
pub async fn get_recent_performance_items_for(
    pool: &DbPool,
    account_id: &str,
    limit: u32,
) -> Result<Vec<PerformanceItem>, StorageError> {
    let rows: Vec<PerformanceRow> = sqlx::query_as(
        "SELECT 'reply' as content_type, \
                SUBSTR(rs.reply_content, 1, 120) as content_preview, \
                rp.likes_received, rp.replies_received, 0 as retweets, \
                rp.impressions, rp.performance_score, rs.created_at as posted_at \
         FROM reply_performance rp \
         JOIN replies_sent rs ON rs.reply_tweet_id = rp.reply_id \
         WHERE rp.account_id = ? \
         UNION ALL \
         SELECT 'tweet' as content_type, \
                SUBSTR(ot.content, 1, 120) as content_preview, \
                tp.likes_received, tp.replies_received, tp.retweets_received, \
                tp.impressions, tp.performance_score, ot.created_at as posted_at \
         FROM tweet_performance tp \
         JOIN original_tweets ot ON ot.tweet_id = tp.tweet_id \
         WHERE tp.account_id = ? \
         ORDER BY posted_at DESC \
         LIMIT ?",
    )
    .bind(account_id)
    .bind(account_id)
    .bind(limit)
    .fetch_all(pool)
    .await
    .map_err(|e| StorageError::Query { source: e })?;

    Ok(rows
        .into_iter()
        .map(|r| PerformanceItem {
            content_type: r.0,
            content_preview: r.1,
            likes: r.2,
            replies_received: r.3,
            retweets: r.4,
            impressions: r.5,
            performance_score: r.6,
            posted_at: r.7,
        })
        .collect())
}

/// Get recent content performance items, newest first.
///
/// Unions reply and tweet performance joined with their content tables
/// so the dashboard can show a content preview alongside metrics.
pub async fn get_recent_performance_items(
    pool: &DbPool,
    limit: u32,
) -> Result<Vec<PerformanceItem>, StorageError> {
    get_recent_performance_items_for(pool, DEFAULT_ACCOUNT_ID, limit).await
}

/// Hourly posting performance data.
#[derive(Debug, Clone, serde::Serialize)]
pub struct HourlyPerformance {
    /// Hour of day (0-23).
    pub hour: i64,
    /// Average engagement score for posts in this hour.
    pub avg_engagement: f64,
    /// Number of posts in this hour.
    pub post_count: i64,
}

/// Get optimal posting times based on historical performance for a specific account.
pub async fn get_optimal_posting_times_for(
    pool: &DbPool,
    account_id: &str,
) -> Result<Vec<HourlyPerformance>, StorageError> {
    let rows: Vec<(i64, f64, i64)> = sqlx::query_as(
        "SELECT
            CAST(strftime('%H', ot.created_at) AS INTEGER) as hour,
            COALESCE(AVG(tp.performance_score), 0.0) as avg_engagement,
            COUNT(*) as post_count
         FROM original_tweets ot
         LEFT JOIN tweet_performance tp ON tp.tweet_id = ot.tweet_id
         WHERE ot.account_id = ? AND ot.status = 'sent' AND ot.tweet_id IS NOT NULL
         GROUP BY hour
         ORDER BY avg_engagement DESC",
    )
    .bind(account_id)
    .fetch_all(pool)
    .await
    .map_err(|e| StorageError::Query { source: e })?;

    Ok(rows
        .into_iter()
        .map(|(hour, avg_engagement, post_count)| HourlyPerformance {
            hour,
            avg_engagement,
            post_count,
        })
        .collect())
}

/// Get optimal posting times based on historical performance.
pub async fn get_optimal_posting_times(
    pool: &DbPool,
) -> Result<Vec<HourlyPerformance>, StorageError> {
    get_optimal_posting_times_for(pool, DEFAULT_ACCOUNT_ID).await
}

// ============================================================================
// Archetype & engagement score updates (Winning DNA pipeline)
// ============================================================================

/// Update the archetype_vibe classification for a tweet performance record.
pub async fn update_tweet_archetype(
    pool: &DbPool,
    tweet_id: &str,
    archetype_vibe: &str,
) -> Result<(), StorageError> {
    sqlx::query("UPDATE tweet_performance SET archetype_vibe = ? WHERE tweet_id = ?")
        .bind(archetype_vibe)
        .bind(tweet_id)
        .execute(pool)
        .await
        .map_err(|e| StorageError::Query { source: e })?;
    Ok(())
}

/// Update the archetype_vibe classification for a reply performance record.
pub async fn update_reply_archetype(
    pool: &DbPool,
    reply_id: &str,
    archetype_vibe: &str,
) -> Result<(), StorageError> {
    sqlx::query("UPDATE reply_performance SET archetype_vibe = ? WHERE reply_id = ?")
        .bind(archetype_vibe)
        .bind(reply_id)
        .execute(pool)
        .await
        .map_err(|e| StorageError::Query { source: e })?;
    Ok(())
}

/// Update the engagement_score for a tweet performance record.
pub async fn update_tweet_engagement_score(
    pool: &DbPool,
    tweet_id: &str,
    score: f64,
) -> Result<(), StorageError> {
    sqlx::query("UPDATE tweet_performance SET engagement_score = ? WHERE tweet_id = ?")
        .bind(score)
        .bind(tweet_id)
        .execute(pool)
        .await
        .map_err(|e| StorageError::Query { source: e })?;
    Ok(())
}

/// Update the engagement_score for a reply performance record.
pub async fn update_reply_engagement_score(
    pool: &DbPool,
    reply_id: &str,
    score: f64,
) -> Result<(), StorageError> {
    sqlx::query("UPDATE reply_performance SET engagement_score = ? WHERE reply_id = ?")
        .bind(score)
        .bind(reply_id)
        .execute(pool)
        .await
        .map_err(|e| StorageError::Query { source: e })?;
    Ok(())
}

/// Get the maximum performance_score across all tweets and replies.
///
/// Returns 0.0 if no performance data exists. Used to normalize engagement scores.
pub async fn get_max_performance_score(pool: &DbPool) -> Result<f64, StorageError> {
    let row: (f64,) = sqlx::query_as(
        "SELECT COALESCE(MAX(max_score), 0.0) FROM (\
             SELECT MAX(performance_score) as max_score FROM tweet_performance \
             UNION ALL \
             SELECT MAX(performance_score) as max_score FROM reply_performance\
         )",
    )
    .fetch_one(pool)
    .await
    .map_err(|e| StorageError::Query { source: e })?;

    Ok(row.0)
}

/// Row type returned by the ancestor retrieval query.
#[derive(Debug, Clone)]
pub struct AncestorRow {
    /// "tweet" or "reply"
    pub content_type: String,
    /// The tweet or reply ID.
    pub id: String,
    /// Truncated content preview (up to 120 chars).
    pub content_preview: String,
    /// Archetype classification (may be None if not yet classified).
    pub archetype_vibe: Option<String>,
    /// Normalized engagement score (0.0-1.0).
    pub engagement_score: Option<f64>,
    /// Raw performance score.
    pub performance_score: f64,
    /// When the content was posted (ISO-8601).
    pub posted_at: String,
}

/// Row type returned by the ancestor retrieval UNION query.
type AncestorQueryRow = (
    String,
    String,
    String,
    Option<String>,
    Option<f64>,
    f64,
    String,
);

/// Convert an ancestor query row tuple into an `AncestorRow` struct.
fn ancestor_row_from_tuple(r: AncestorQueryRow) -> AncestorRow {
    AncestorRow {
        content_type: r.0,
        id: r.1,
        content_preview: r.2,
        archetype_vibe: r.3,
        engagement_score: r.4,
        performance_score: r.5,
        posted_at: r.6,
    }
}

/// Query scored ancestors with engagement_score populated.
///
/// Returns ancestors where `engagement_score >= min_score`, ordered by
/// engagement_score DESC. For topic matching, uses the `topic` column on
/// original_tweets and LIKE-based content matching on replies.
pub async fn get_scored_ancestors(
    pool: &DbPool,
    topic_keywords: &[String],
    min_score: f64,
    limit: u32,
) -> Result<Vec<AncestorRow>, StorageError> {
    if topic_keywords.is_empty() {
        // No keywords: return top ancestors regardless of topic
        let rows: Vec<AncestorQueryRow> = sqlx::query_as(
            "SELECT 'tweet' as content_type, tp.tweet_id, \
                        SUBSTR(ot.content, 1, 120), \
                        tp.archetype_vibe, tp.engagement_score, tp.performance_score, \
                        ot.created_at \
                 FROM tweet_performance tp \
                 JOIN original_tweets ot ON ot.tweet_id = tp.tweet_id \
                 WHERE tp.engagement_score IS NOT NULL \
                   AND tp.engagement_score >= ? \
                 UNION ALL \
                 SELECT 'reply', rp.reply_id, SUBSTR(rs.reply_content, 1, 120), \
                        rp.archetype_vibe, rp.engagement_score, rp.performance_score, \
                        rs.created_at \
                 FROM reply_performance rp \
                 JOIN replies_sent rs ON rs.reply_tweet_id = rp.reply_id \
                 WHERE rp.engagement_score IS NOT NULL \
                   AND rp.engagement_score >= ? \
                 ORDER BY engagement_score DESC \
                 LIMIT ?",
        )
        .bind(min_score)
        .bind(min_score)
        .bind(limit)
        .fetch_all(pool)
        .await
        .map_err(|e| StorageError::Query { source: e })?;

        return Ok(rows.into_iter().map(ancestor_row_from_tuple).collect());
    }

    // Build parameterized IN clause for tweet topics and LIKE clauses for replies.
    // SQLx uses sequential `?` placeholders for SQLite.
    let topic_placeholders: String = (0..topic_keywords.len())
        .map(|_| "?".to_string())
        .collect::<Vec<_>>()
        .join(", ");

    let like_conditions: Vec<String> = (0..topic_keywords.len())
        .map(|_| "rs.reply_content LIKE '%' || ? || '%'".to_string())
        .collect();
    let like_clause = like_conditions.join(" OR ");

    let query_str = format!(
        "SELECT 'tweet' as content_type, tp.tweet_id, \
                SUBSTR(ot.content, 1, 120), \
                tp.archetype_vibe, tp.engagement_score, tp.performance_score, \
                ot.created_at \
         FROM tweet_performance tp \
         JOIN original_tweets ot ON ot.tweet_id = tp.tweet_id \
         WHERE tp.engagement_score IS NOT NULL \
           AND tp.engagement_score >= ? \
           AND (ot.topic IN ({topic_placeholders})) \
         UNION ALL \
         SELECT 'reply', rp.reply_id, SUBSTR(rs.reply_content, 1, 120), \
                rp.archetype_vibe, rp.engagement_score, rp.performance_score, \
                rs.created_at \
         FROM reply_performance rp \
         JOIN replies_sent rs ON rs.reply_tweet_id = rp.reply_id \
         WHERE rp.engagement_score IS NOT NULL \
           AND rp.engagement_score >= ? \
           AND ({like_clause}) \
         ORDER BY engagement_score DESC \
         LIMIT ?"
    );

    let mut query = sqlx::query_as::<_, AncestorQueryRow>(&query_str);

    // Bind: min_score for tweets, then topic keywords for IN clause
    query = query.bind(min_score);
    for kw in topic_keywords {
        query = query.bind(kw);
    }
    // Bind: min_score for replies, then keywords for LIKE clauses, then limit
    query = query.bind(min_score);
    for kw in topic_keywords {
        query = query.bind(kw);
    }
    query = query.bind(limit);

    let rows = query
        .fetch_all(pool)
        .await
        .map_err(|e| StorageError::Query { source: e })?;

    Ok(rows.into_iter().map(ancestor_row_from_tuple).collect())
}

#[cfg(test)]
mod tests {
    use super::*;
    use crate::storage::init_test_db;

    #[tokio::test]
    async fn upsert_and_get_follower_snapshot() {
        let pool = init_test_db().await.expect("init db");

        upsert_follower_snapshot(&pool, 1000, 200, 500)
            .await
            .expect("upsert");

        let snapshots = get_follower_snapshots(&pool, 10).await.expect("get");
        assert_eq!(snapshots.len(), 1);
        assert_eq!(snapshots[0].follower_count, 1000);
        assert_eq!(snapshots[0].following_count, 200);
        assert_eq!(snapshots[0].tweet_count, 500);
    }

    #[tokio::test]
    async fn upsert_follower_snapshot_updates_existing() {
        let pool = init_test_db().await.expect("init db");

        upsert_follower_snapshot(&pool, 1000, 200, 500)
            .await
            .expect("upsert");
        upsert_follower_snapshot(&pool, 1050, 201, 510)
            .await
            .expect("upsert again");

        let snapshots = get_follower_snapshots(&pool, 10).await.expect("get");
        assert_eq!(snapshots.len(), 1);
        assert_eq!(snapshots[0].follower_count, 1050);
    }

    #[tokio::test]
    async fn upsert_reply_performance_works() {
        let pool = init_test_db().await.expect("init db");

        upsert_reply_performance(&pool, "r1", 5, 2, 100, 55.0)
            .await
            .expect("upsert");

        // Update
        upsert_reply_performance(&pool, "r1", 10, 3, 200, 75.0)
            .await
            .expect("update");
    }

    #[tokio::test]
    async fn upsert_tweet_performance_works() {
        let pool = init_test_db().await.expect("init db");

        upsert_tweet_performance(&pool, "tw1", 10, 5, 3, 500, 82.0)
            .await
            .expect("upsert");

        // Update
        upsert_tweet_performance(&pool, "tw1", 20, 10, 5, 1000, 95.0)
            .await
            .expect("update");
    }

    #[tokio::test]
    async fn update_and_get_content_scores() {
        let pool = init_test_db().await.expect("init db");

        update_content_score(&pool, "rust", "tip", 80.0)
            .await
            .expect("update");
        update_content_score(&pool, "rust", "tip", 90.0)
            .await
            .expect("update");
        update_content_score(&pool, "python", "list", 60.0)
            .await
            .expect("update");

        let top = get_top_topics(&pool, 10).await.expect("get");
        assert_eq!(top.len(), 2);
        // Rust should be higher (avg ~85) than Python (60)
        assert_eq!(top[0].topic, "rust");
        assert_eq!(top[0].total_posts, 2);
        assert!(top[0].avg_performance > 80.0);
    }

    #[test]
    fn compute_performance_score_basic() {
        let score = compute_performance_score(10, 5, 3, 1000);
        // (10*3 + 5*5 + 3*4) / 1000 * 1000 = (30 + 25 + 12) = 67
        assert!((score - 67.0).abs() < 0.01);
    }

    #[test]
    fn compute_performance_score_zero_impressions() {
        let score = compute_performance_score(10, 5, 3, 0);
        // Denominator clamped to 1: (30 + 25 + 12) / 1 * 1000 = 67000
        assert!((score - 67000.0).abs() < 0.01);
    }

    #[test]
    fn compute_performance_score_all_zero() {
        let score = compute_performance_score(0, 0, 0, 0);
        assert!((score - 0.0).abs() < 0.01);
    }

    #[tokio::test]
    async fn avg_reply_engagement_empty() {
        let pool = init_test_db().await.expect("init db");
        let avg = get_avg_reply_engagement(&pool).await.expect("avg");
        assert!((avg - 0.0).abs() < 0.01);
    }

    #[tokio::test]
    async fn avg_reply_engagement_with_data() {
        let pool = init_test_db().await.expect("init db");
        upsert_reply_performance(&pool, "r1", 10, 5, 1000, 67.0)
            .await
            .expect("upsert");
        upsert_reply_performance(&pool, "r2", 20, 10, 2000, 80.0)
            .await
            .expect("upsert");

        let avg = get_avg_reply_engagement(&pool).await.expect("avg");
        // (67 + 80) / 2 = 73.5
        assert!((avg - 73.5).abs() < 0.01);
    }

    #[tokio::test]
    async fn avg_tweet_engagement_empty() {
        let pool = init_test_db().await.expect("init db");
        let avg = get_avg_tweet_engagement(&pool).await.expect("avg");
        assert!((avg - 0.0).abs() < 0.01);
    }

    #[tokio::test]
    async fn performance_counts_empty() {
        let pool = init_test_db().await.expect("init db");
        let (replies, tweets) = get_performance_counts(&pool).await.expect("counts");
        assert_eq!(replies, 0);
        assert_eq!(tweets, 0);
    }

    #[tokio::test]
    async fn performance_counts_with_data() {
        let pool = init_test_db().await.expect("init db");
        upsert_reply_performance(&pool, "r1", 10, 5, 1000, 67.0)
            .await
            .expect("upsert");
        upsert_tweet_performance(&pool, "tw1", 10, 5, 3, 500, 82.0)
            .await
            .expect("upsert");
        upsert_tweet_performance(&pool, "tw2", 20, 10, 5, 1000, 95.0)
            .await
            .expect("upsert");

        let (replies, tweets) = get_performance_counts(&pool).await.expect("counts");
        assert_eq!(replies, 1);
        assert_eq!(tweets, 2);
    }

    #[tokio::test]
    async fn analytics_summary_empty() {
        let pool = init_test_db().await.expect("init db");
        let summary = get_analytics_summary(&pool).await.expect("summary");
        assert_eq!(summary.followers.current, 0);
        assert_eq!(summary.followers.change_7d, 0);
        assert_eq!(summary.followers.change_30d, 0);
        assert_eq!(summary.actions_today.replies, 0);
        assert!((summary.engagement.avg_reply_score - 0.0).abs() < 0.01);
        assert!(summary.top_topics.is_empty());
    }

    #[tokio::test]
    async fn analytics_summary_with_data() {
        let pool = init_test_db().await.expect("init db");

        // Insert follower snapshot (only today since test db is in-memory)
        upsert_follower_snapshot(&pool, 1000, 200, 500)
            .await
            .expect("upsert");

        // Insert some performance data
        upsert_reply_performance(&pool, "r1", 10, 5, 1000, 67.0)
            .await
            .expect("upsert");

        // Insert content scores so top_topics is populated
        update_content_score(&pool, "rust", "tip", 80.0)
            .await
            .expect("score");
        update_content_score(&pool, "ai", "thread", 60.0)
            .await
            .expect("score");

        let summary = get_analytics_summary(&pool).await.expect("summary");
        assert_eq!(summary.followers.current, 1000);
        assert!(summary.engagement.avg_reply_score > 0.0);
        assert_eq!(summary.engagement.total_replies_sent, 1);
        assert_eq!(summary.top_topics.len(), 2);
        assert_eq!(summary.top_topics[0].topic, "rust");
    }

    #[tokio::test]
    async fn recent_performance_items_empty() {
        let pool = init_test_db().await.expect("init db");
        let items = get_recent_performance_items(&pool, 10).await.expect("get");
        assert!(items.is_empty());
    }

    #[tokio::test]
    async fn recent_performance_items_with_data() {
        let pool = init_test_db().await.expect("init db");

        // Insert a reply and its performance
        let reply = crate::storage::replies::ReplySent {
            id: 0,
            target_tweet_id: "t1".to_string(),
            reply_tweet_id: Some("r1".to_string()),
            reply_content: "Great point about testing!".to_string(),
            llm_provider: Some("openai".to_string()),
            llm_model: Some("gpt-4o".to_string()),
            created_at: "2026-02-23T12:00:00Z".to_string(),
            status: "sent".to_string(),
            error_message: None,
        };
        crate::storage::replies::insert_reply(&pool, &reply)
            .await
            .expect("insert reply");
        upsert_reply_performance(&pool, "r1", 10, 5, 1000, 67.0)
            .await
            .expect("upsert perf");

        let items = get_recent_performance_items(&pool, 10).await.expect("get");
        assert_eq!(items.len(), 1);
        assert_eq!(items[0].content_type, "reply");
        assert!(items[0].content_preview.contains("testing"));
        assert_eq!(items[0].likes, 10);
    }

    // ============================================================================
    // Winning DNA storage tests
    // ============================================================================

    #[tokio::test]
    async fn update_and_get_tweet_archetype() {
        let pool = init_test_db().await.expect("init db");

        upsert_tweet_performance(&pool, "tw1", 10, 5, 3, 500, 82.0)
            .await
            .expect("upsert");

        update_tweet_archetype(&pool, "tw1", "list")
            .await
            .expect("update");

        let row: (Option<String>,) =
            sqlx::query_as("SELECT archetype_vibe FROM tweet_performance WHERE tweet_id = ?")
                .bind("tw1")
                .fetch_one(&pool)
                .await
                .expect("query");
        assert_eq!(row.0.as_deref(), Some("list"));
    }

    #[tokio::test]
    async fn update_and_get_reply_archetype() {
        let pool = init_test_db().await.expect("init db");

        upsert_reply_performance(&pool, "r1", 10, 5, 1000, 67.0)
            .await
            .expect("upsert");

        update_reply_archetype(&pool, "r1", "ask_question")
            .await
            .expect("update");

        let row: (Option<String>,) =
            sqlx::query_as("SELECT archetype_vibe FROM reply_performance WHERE reply_id = ?")
                .bind("r1")
                .fetch_one(&pool)
                .await
                .expect("query");
        assert_eq!(row.0.as_deref(), Some("ask_question"));
    }

    #[tokio::test]
    async fn update_and_get_engagement_score() {
        let pool = init_test_db().await.expect("init db");

        upsert_tweet_performance(&pool, "tw1", 10, 5, 3, 500, 82.0)
            .await
            .expect("upsert");

        update_tweet_engagement_score(&pool, "tw1", 0.85)
            .await
            .expect("update");

        let row: (Option<f64>,) =
            sqlx::query_as("SELECT engagement_score FROM tweet_performance WHERE tweet_id = ?")
                .bind("tw1")
                .fetch_one(&pool)
                .await
                .expect("query");
        assert!((row.0.unwrap() - 0.85).abs() < 0.001);
    }

    #[tokio::test]
    async fn get_max_performance_score_empty() {
        let pool = init_test_db().await.expect("init db");
        let max = get_max_performance_score(&pool).await.expect("max");
        assert!((max - 0.0).abs() < 0.01);
    }

    #[tokio::test]
    async fn get_max_performance_score_with_data() {
        let pool = init_test_db().await.expect("init db");

        upsert_tweet_performance(&pool, "tw1", 10, 5, 3, 500, 82.0)
            .await
            .expect("upsert");
        upsert_reply_performance(&pool, "r1", 20, 10, 2000, 95.0)
            .await
            .expect("upsert");

        let max = get_max_performance_score(&pool).await.expect("max");
        assert!((max - 95.0).abs() < 0.01);
    }

    #[tokio::test]
    async fn get_scored_ancestors_empty() {
        let pool = init_test_db().await.expect("init db");
        let ancestors = get_scored_ancestors(&pool, &[], 0.1, 10)
            .await
            .expect("query");
        assert!(ancestors.is_empty());
    }

    #[tokio::test]
    async fn get_scored_ancestors_returns_scored_items() {
        let pool = init_test_db().await.expect("init db");

        // Insert a tweet with performance + engagement_score
        sqlx::query(
            "INSERT INTO original_tweets (account_id, tweet_id, content, topic, status, created_at) \
             VALUES ('00000000-0000-0000-0000-000000000000', 'tw1', 'Great Rust testing tips', 'rust', 'sent', '2026-02-27T10:00:00Z')",
        )
        .execute(&pool)
        .await
        .expect("insert tweet");

        upsert_tweet_performance(&pool, "tw1", 10, 5, 3, 500, 82.0)
            .await
            .expect("upsert perf");
        update_tweet_engagement_score(&pool, "tw1", 0.85)
            .await
            .expect("update score");

        let ancestors = get_scored_ancestors(&pool, &[], 0.1, 10)
            .await
            .expect("query");
        assert_eq!(ancestors.len(), 1);
        assert_eq!(ancestors[0].content_type, "tweet");
        assert_eq!(ancestors[0].id, "tw1");
        assert!((ancestors[0].engagement_score.unwrap() - 0.85).abs() < 0.001);
    }

    #[tokio::test]
    async fn get_scored_ancestors_filters_low_engagement() {
        let pool = init_test_db().await.expect("init db");

        sqlx::query(
            "INSERT INTO original_tweets (account_id, tweet_id, content, topic, status, created_at) \
             VALUES ('00000000-0000-0000-0000-000000000000', 'tw1', 'Low performer', 'rust', 'sent', '2026-02-27T10:00:00Z')",
        )
        .execute(&pool)
        .await
        .expect("insert tweet");

        upsert_tweet_performance(&pool, "tw1", 1, 0, 0, 500, 5.0)
            .await
            .expect("upsert perf");
        update_tweet_engagement_score(&pool, "tw1", 0.05)
            .await
            .expect("update score");

        // min_score = 0.1, so this ancestor should be filtered out
        let ancestors = get_scored_ancestors(&pool, &[], 0.1, 10)
            .await
            .expect("query");
        assert!(ancestors.is_empty());
    }
}