`agents/twitter-agent/skills/monitor.js`

241 lines
/**
 * Twitter Monitor Skill
 *
 * Monitors configured hashtags and keywords from sdcc_agent_monitors table.
 * Fetches recent tweets using Twitter API v2 search.
 *
 * Rate limit: uses search limit (60/15min)
 */

const { RateLimiter } = require('../../shared/rate-limiter');
const { logAction } = require('../../shared/audit-logger');
const { pushSocialFeed, pushTrending } = require('../../shared/pulse-reporter');
const { searchTweets, ADVOCACY_HASHTAGS } = require('../lib/twitter');
const { query: dbQuery } = require('../../shared/db');

const AGENT = 'twitter-agent';
const PLATFORM = 'twitter';

const rateLimiter = new RateLimiter(AGENT, PLATFORM);

/**
 * @param {Object} body - Request body
 * @param {string[]} [body.hashtags] - Override hashtags to monitor
 * @param {string[]} [body.keywords] - Additional keywords to search
 * @param {number} [body.max_results=10] - Max results per query
 * @param {boolean} [body.push_to_pulse=true] - Push results to Pulse
 * @returns {Promise<Object>}
 */
module.exports = async function monitor(body) {
  const maxResults = Math.min(body.max_results || 10, 100);
  const pushToPulse = body.push_to_pulse !== false;

  // Check rate limit
  const limit = await rateLimiter.checkLimit('search');
  if (!limit.allowed) {
    await logAction({
      agent: AGENT,
      actionType: 'monitor',
      platform: PLATFORM,
      status: 'rate_limited',
      errorMessage: `Rate limited. Retry after ${Math.ceil(limit.retryAfterMs / 1000)}s`,
    });

    return {
      rate_limited: true,
      retry_after_seconds: Math.ceil(limit.retryAfterMs / 1000),
      current: limit.current,
      max: limit.max,
    };
  }

  // Get active monitors from database
  let monitors = [];
  try {
    const { rows } = await dbQuery(
      `SELECT id, target, keywords, monitor_type
       FROM sdcc_agent_monitors
       WHERE agent = $1 AND platform = $2 AND is_active = true`,
      [AGENT, PLATFORM]
    );
    monitors = rows;
  } catch (err) {
    console.error(`[${AGENT}] Failed to fetch monitors:`, err.message);
  }

  // Build search queries from monitors and/or body params
  const queries = [];

  // Add monitor-based queries
  for (const m of monitors) {
    if (m.monitor_type === 'hashtag') {
      queries.push({
        query: `${m.target} -is:retweet lang:en`,
        monitor_id: m.id,
        source: 'db_monitor',
      });
    } else if (m.monitor_type === 'keyword' && m.keywords) {
      queries.push({
        query: `(${m.keywords.join(' OR ')}) -is:retweet lang:en`,
        monitor_id: m.id,
        source: 'db_monitor',
      });
    }
  }

  // Add body-specified queries
  if (body.hashtags && body.hashtags.length > 0) {
    queries.push({
      query: `(${body.hashtags.join(' OR ')}) -is:retweet lang:en`,
      source: 'request',
    });
  }

  if (body.keywords && body.keywords.length > 0) {
    queries.push({
      query: `(${body.keywords.join(' OR ')}) -is:retweet lang:en`,
      source: 'request',
    });
  }

  // Fallback: use default hashtags if no queries configured
  if (queries.length === 0) {
    queries.push({
      query: `(${ADVOCACY_HASHTAGS.slice(0, 3).join(' OR ')}) -is:retweet lang:en`,
      source: 'default',
    });
  }

  // Execute searches
  const allResults = [];
  for (const q of queries) {
    try {
      const searchResult = await searchTweets(q.query, maxResults);
      await rateLimiter.recordAction('search');

      const tweets = searchResult.tweets || [];

      // Compute engagement score for each tweet
      const scored = tweets.map((t) => {
        const m = t.metrics || {};
        const engagement =
          (m.like_count || 0) * 1 +
          (m.retweet_count || 0) * 2 +
          (m.reply_count || 0) * 1.5 +
          (m.quote_count || 0) * 2;
        return { ...t, engagement_score: engagement };
      });

      // Sort by engagement
      scored.sort((a, b) => b.engagement_score - a.engagement_score);

      allResults.push({
        query: q.query,
        source: q.source,
        monitor_id: q.monitor_id || null,
        count: scored.length,
        tweets: scored,
        simulated: searchResult.simulated || false,
      });

      // Update monitor last_checked_at and items_found
      if (q.monitor_id) {
        try {
          await dbQuery(
            `UPDATE sdcc_agent_monitors
             SET last_checked_at = NOW(), items_found = items_found + $1, updated_at = NOW()
             WHERE id = $2`,
            [scored.length, q.monitor_id]
          );
        } catch (err) {
          console.error(`[${AGENT}] Failed to update monitor ${q.monitor_id}:`, err.message);
        }
      }
    } catch (err) {
      console.error(`[${AGENT}] Search failed for query "${q.query}":`, err.message);
      allResults.push({
        query: q.query,
        source: q.source,
        count: 0,
        tweets: [],
        error: err.message,
      });
    }
  }

  // Aggregate trending hashtags across all results
  const hashtagCounts = {};
  allResults.forEach((r) => {
    r.tweets.forEach((t) => {
      const matches = t.text.match(/#\w+/g) || [];
      matches.forEach((tag) => {
        const lower = tag.toLowerCase();
        hashtagCounts[lower] = (hashtagCounts[lower] || 0) + 1;
      });
    });
  });

  const trending = Object.entries(hashtagCounts)
    .sort(([, a], [, b]) => b - a)
    .slice(0, 15)
    .map(([tag, count]) => ({
      topic: tag,
      platform: PLATFORM,
      volume: count,
      sentiment: 'neutral',
    }));

  // Push to Pulse
  if (pushToPulse) {
    if (trending.length > 0) {
      await pushTrending(trending);
    }

    // Push top-engagement tweets
    const topTweets = allResults
      .flatMap((r) => r.tweets)
      .sort((a, b) => b.engagement_score - a.engagement_score)
      .slice(0, 5);

    for (const tweet of topTweets) {
      await pushSocialFeed({
        agent: AGENT,
        platform: PLATFORM,
        type: 'mention',
        title: `@${tweet.author?.username || 'unknown'}: ${tweet.text.substring(0, 80)}`,
        url: `https://x.com/${tweet.author?.username || 'i'}/status/${tweet.id}`,
        content: tweet.text,
        metadata: {
          tweet_id: tweet.id,
          engagement_score: tweet.engagement_score,
          metrics: tweet.metrics,
        },
      });
    }
  }

  const totalTweets = allResults.reduce((sum, r) => sum + r.count, 0);

  // Log the action
  await logAction({
    agent: AGENT,
    actionType: 'monitor',
    platform: PLATFORM,
    responseData: {
      queries_executed: allResults.length,
      total_tweets: totalTweets,
      trending_hashtags: trending.length,
      monitors_checked: monitors.length,
    },
    status: 'success',
  });

  return {
    queries_executed: allResults.length,
    total_tweets: totalTweets,
    results: allResults,
    trending,
    monitored_at: new Date().toISOString(),
  };
};