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

199 lines
/**
 * Discord Monitor Skill
 *
 * Manages monitoring configuration for Discord channels and keywords.
 * Stores/updates monitor entries in the sdcc_agent_monitors table.
 * Returns current monitor configuration.
 *
 * Note: Active real-time monitoring is event-driven via Discord.js client
 * events in the bot, not cron-based. This skill manages the configuration.
 */

const { logAction } = require('../../shared/audit-logger');
const { getRecentMessages, scoreRelevance, hasCredentials, KEYWORDS } = require('../lib/discord');
const { pushSocialFeed } = require('../../shared/pulse-reporter');
const { query: dbQuery } = require('../../shared/db');

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

/**
 * @param {Object} body - Request body
 * @param {string[]} [body.channel_ids] - Channel IDs to add/update as monitors
 * @param {string[]} [body.keywords] - Keywords to monitor for
 * @param {boolean} [body.check_now=false] - If true, fetch recent messages from monitored channels now
 * @param {number} [body.limit=50] - Messages per channel when checking
 * @param {boolean} [body.list_only=false] - If true, just return current monitors
 * @returns {Promise<Object>}
 */
module.exports = async function monitor(body) {
  const checkNow = body.check_now || false;
  const listOnly = body.list_only || false;
  const limit = Math.min(body.limit || 50, 100);

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

  // If list_only, just return current state
  if (listOnly) {
    return {
      monitors: currentMonitors,
      total: currentMonitors.length,
      active: currentMonitors.filter((m) => m.is_active).length,
      bot_connected: hasCredentials(),
    };
  }

  // Add/update channel monitors if provided
  const addedMonitors = [];
  if (body.channel_ids && body.channel_ids.length > 0) {
    const keywords = body.keywords || KEYWORDS.slice(0, 5);

    for (const channelId of body.channel_ids) {
      try {
        const { rows } = await dbQuery(
          `INSERT INTO sdcc_agent_monitors
            (agent, platform, target, monitor_type, keywords, is_active)
           VALUES ($1, $2, $3, 'channel', $4, true)
           ON CONFLICT (agent, platform, target)
           DO UPDATE SET keywords = $4, is_active = true, updated_at = NOW()
           RETURNING *`,
          [AGENT, PLATFORM, channelId, JSON.stringify(keywords)]
        );
        if (rows[0]) {
          addedMonitors.push(rows[0]);
        }
      } catch (err) {
        console.error(`[${AGENT}] Failed to upsert monitor for channel ${channelId}:`, err.message);
        addedMonitors.push({
          target: channelId,
          error: err.message,
        });
      }
    }
  }

  // Check monitored channels now if requested
  let checkResults = [];
  if (checkNow) {
    const activeMonitors = currentMonitors.filter((m) => m.is_active);
    const channelsToCheck = addedMonitors.length > 0
      ? addedMonitors.filter((m) => m.id).map((m) => m.target)
      : activeMonitors.map((m) => m.target);

    for (const channelId of channelsToCheck) {
      try {
        const messages = await getRecentMessages(channelId, limit);

        // Score and filter relevant messages
        const scored = messages
          .filter((msg) => !msg.isBot)
          .map((msg) => ({
            ...msg,
            relevance_score: scoreRelevance(msg.content),
          }));

        const relevant = scored.filter((m) => m.relevance_score > 10);

        checkResults.push({
          channelId,
          total_messages: messages.length,
          relevant_count: relevant.length,
          top_messages: relevant.slice(0, 5),
        });

        // Update monitor stats
        const mon = [...currentMonitors, ...addedMonitors].find((m) => m.target === channelId);
        if (mon && mon.id) {
          await dbQuery(
            `UPDATE sdcc_agent_monitors
             SET last_checked_at = NOW(), items_found = items_found + $1, updated_at = NOW()
             WHERE id = $2`,
            [relevant.length, mon.id]
          );
        }

        // Push high-relevance messages to Pulse
        for (const msg of relevant.slice(0, 3)) {
          await pushSocialFeed({
            agent: AGENT,
            platform: PLATFORM,
            type: 'discussion',
            title: `${msg.author}: ${msg.content.substring(0, 80)}`,
            content: msg.content.substring(0, 300),
            metadata: {
              message_id: msg.id,
              channel_id: channelId,
              author: msg.author,
              relevance_score: msg.relevance_score,
            },
          });
        }
      } catch (err) {
        console.error(`[${AGENT}] Check failed for channel ${channelId}:`, err.message);
        checkResults.push({
          channelId,
          total_messages: 0,
          relevant_count: 0,
          error: err.message,
        });
      }
    }
  }

  // Refresh monitor list after changes
  let updatedMonitors = currentMonitors;
  if (addedMonitors.length > 0) {
    try {
      const { rows } = await dbQuery(
        `SELECT id, target, keywords, monitor_type, is_active,
                items_found, last_checked_at, created_at
         FROM sdcc_agent_monitors
         WHERE agent = $1 AND platform = $2
         ORDER BY created_at DESC`,
        [AGENT, PLATFORM]
      );
      updatedMonitors = rows;
    } catch (err) {
      console.error(`[${AGENT}] Failed to refresh monitors:`, err.message);
    }
  }

  // Log the action
  await logAction({
    agent: AGENT,
    actionType: 'monitor',
    platform: PLATFORM,
    responseData: {
      monitors_total: updatedMonitors.length,
      monitors_active: updatedMonitors.filter((m) => m.is_active).length,
      channels_added: addedMonitors.length,
      channels_checked: checkResults.length,
    },
    status: 'success',
  });

  return {
    monitors: updatedMonitors,
    total: updatedMonitors.length,
    active: updatedMonitors.filter((m) => m.is_active).length,
    added: addedMonitors.length > 0 ? addedMonitors : undefined,
    check_results: checkResults.length > 0 ? checkResults : undefined,
    bot_connected: hasCredentials(),
    monitored_at: new Date().toISOString(),
  };
};