← back to Norma

agents/moveon-agent/skills/monitor.js

161 lines

/**
 * MoveOn Monitor Skill
 *
 * Tracks signature counts on live petitions.
 * Can monitor a specific petition URL or all pipeline items with platform=moveon.
 *
 * Rate limit: 2/hour
 */

const { RateLimiter } = require('../../shared/rate-limiter');
const { logAction } = require('../../shared/audit-logger');
const { getPetitionDetails } = require('../lib/moveon');
const { query: dbQuery } = require('../../shared/db');

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

const rateLimiter = new RateLimiter(AGENT, PLATFORM);

/**
 * @param {Object} body - Request body
 * @param {string} [body.petition_url] - Specific petition URL to monitor
 * @param {boolean} [body.check_all=false] - Check all pipeline items with platform=moveon
 * @returns {Promise<Object>}
 */
module.exports = async function monitor(body) {
  // Check rate limit
  const limit = await rateLimiter.checkLimit('monitor');
  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,
    };
  }

  const results = [];

  if (body.petition_url) {
    // Monitor a single petition
    try {
      const details = await getPetitionDetails(body.petition_url);
      results.push({
        url: body.petition_url,
        title: details.title,
        signatures: details.signatures,
        goal: details.goal,
        status: 'tracked',
      });

      // Try to update the pipeline entry if it exists
      await _updatePipelineSignatures(body.petition_url, details.signatures);
    } catch (err) {
      results.push({
        url: body.petition_url,
        status: 'error',
        error: err.message,
      });
    }
  }

  if (body.check_all || !body.petition_url) {
    // Check all pipeline items with platform=moveon that have a URL
    try {
      const { rows: pipelineItems } = await dbQuery(
        `SELECT id, title, platform_url, signature_count, signature_goal
         FROM petition_pipeline
         WHERE platform = 'moveon'
           AND platform_url IS NOT NULL
           AND status IN ('posted', 'active', 'approved')
         ORDER BY created_at DESC
         LIMIT 20`
      );

      for (const item of pipelineItems) {
        try {
          const details = await getPetitionDetails(item.platform_url);
          const previousCount = item.signature_count || 0;
          const newCount = details.signatures;
          const delta = newCount - previousCount;

          // Update the pipeline entry
          await dbQuery(
            `UPDATE petition_pipeline
             SET signature_count = $1, last_tracked_at = NOW(), updated_at = NOW()
             WHERE id = $2`,
            [newCount, item.id]
          );

          results.push({
            pipeline_id: item.id,
            title: item.title,
            url: item.platform_url,
            previous_signatures: previousCount,
            current_signatures: newCount,
            delta,
            goal: details.goal || item.signature_goal,
            status: 'updated',
          });
        } catch (err) {
          results.push({
            pipeline_id: item.id,
            title: item.title,
            url: item.platform_url,
            status: 'error',
            error: err.message,
          });
        }
      }
    } catch (err) {
      console.error(`[${AGENT}] Failed to query pipeline:`, err.message);
    }
  }

  // Record the action
  await rateLimiter.recordAction('monitor');
  await logAction({
    agent: AGENT,
    actionType: 'monitor',
    platform: PLATFORM,
    content: body.petition_url || 'all pipeline items',
    responseData: {
      tracked: results.length,
      updated: results.filter((r) => r.status === 'updated').length,
      errors: results.filter((r) => r.status === 'error').length,
    },
    status: 'success',
  });

  return {
    tracked: results.length,
    petitions: results,
    checked_at: new Date().toISOString(),
  };
};

/**
 * Update signature count in the pipeline for a specific URL.
 */
async function _updatePipelineSignatures(url, signatureCount) {
  try {
    await dbQuery(
      `UPDATE petition_pipeline
       SET signature_count = $1, last_tracked_at = NOW(), updated_at = NOW()
       WHERE platform_url = $2 AND platform = 'moveon'`,
      [signatureCount, url]
    );
  } catch (err) {
    console.error(`[${AGENT}] Failed to update pipeline signatures:`, err.message);
  }
}