smartbotic/nodes/rss/rss-reader.js

/**
 * @node rss-reader
 * @name RSS Reader
 * @category data
 * @version 1.1.0
 * @description Read and parse RSS/Atom feeds with optional filtering by keyword, date, or count. Supports stateful new item detection.
 * @icon rss
 */

const configSchema = {
    type: 'object',
    properties: {
        url: {
            type: 'string',
            title: 'Feed URL',
            description: 'URL of the RSS or Atom feed'
        },
        filterKeyword: {
            type: 'string',
            title: 'Keyword Filter',
            description: 'Filter items by keyword (matches title and description)'
        },
        filterNewerThan: {
            type: 'string',
            title: 'Newer Than',
            description: 'Only include items newer than this date (ISO 8601 format, e.g., 2024-01-15T00:00:00Z)'
        },
        maxItems: {
            type: 'number',
            title: 'Max Items',
            description: 'Maximum number of items to return (0 = unlimited)',
            default: 0
        },
        timeout: {
            type: 'number',
            title: 'Timeout (ms)',
            default: 30000
        },
        detectNewItems: {
            type: 'boolean',
            title: 'Detect New Items',
            description: 'When enabled, only output items that have not been seen in previous executions',
            default: false
        },
        stateCollection: {
            type: 'string',
            title: 'State Collection',
            description: 'Select a read-write collection to store seen item identifiers',
            dynamicOptions: {
                source: 'storage.collections',
                labelField: 'name',
                valueField: 'name',
                filter: { access: 'read-write' }
            },
            showWhen: { field: 'detectNewItems', value: true }
        }
    },
    required: ['url']
};

const inputSchema = {
    type: 'object',
    properties: {
        url: {
            type: 'string',
            description: 'Override feed URL from input'
        },
        filterKeyword: {
            type: 'string',
            description: 'Override keyword filter from input'
        },
        filterNewerThan: {
            type: 'string',
            description: 'Override date filter from input'
        },
        maxItems: {
            type: 'number',
            description: 'Override max items from input'
        }
    }
};

const outputSchema = {
    type: 'object',
    properties: {
        feedTitle: { type: 'string', description: 'Title of the feed' },
        feedLink: { type: 'string', description: 'Link to the feed homepage' },
        feedDescription: { type: 'string', description: 'Description of the feed' },
        itemCount: { type: 'number', description: 'Number of items returned' },
        newItemCount: { type: 'number', description: 'Number of new items (when detectNewItems is enabled)' },
        totalFeedItemCount: { type: 'number', description: 'Total items in feed before filtering (when detectNewItems is enabled)' },
        items: {
            type: 'array',
            description: 'Array of feed items',
            items: {
                type: 'object',
                properties: {
                    title: { type: 'string' },
                    link: { type: 'string' },
                    description: { type: 'string' },
                    pubDate: { type: 'string' },
                    author: { type: 'string' },
                    categories: { type: 'array', items: { type: 'string' } },
                    guid: { type: 'string' }
                }
            }
        }
    }
};

/**
 * Simple XML parser for RSS/Atom feeds
 * Handles basic XML structure without external dependencies
 */
function parseXml() {
    function getTagContent(xml, tagName) {
        const patterns = [
            new RegExp(`<${tagName}[^>]*>([\\s\\S]*?)<\\/${tagName}>`, 'i'),
            new RegExp(`<${tagName}[^>]*\\/>`, 'i')
        ];

        const match = xml.match(patterns[0]);
        if (match) {
            return match[1].trim();
        }
        return null;
    }

    function getAllTagContents(xml, tagName) {
        const results = [];
        const regex = new RegExp(`<${tagName}[^>]*>([\\s\\S]*?)<\\/${tagName}>`, 'gi');
        let match;
        while ((match = regex.exec(xml)) !== null) {
            results.push(match[1].trim());
        }
        return results;
    }

    function getTagAttribute(xml, tagName, attrName) {
        const regex = new RegExp(`<${tagName}[^>]*\\s${attrName}=["']([^"']*)["'][^>]*>`, 'i');
        const match = xml.match(regex);
        return match ? match[1] : null;
    }

    function getAllTags(xml, tagName) {
        const results = [];
        const regex = new RegExp(`<${tagName}[^>]*>([\\s\\S]*?)<\\/${tagName}>|<${tagName}([^>]*)\\/>`, 'gi');
        let match;
        while ((match = regex.exec(xml)) !== null) {
            results.push({
                content: match[1] ? match[1].trim() : '',
                fullMatch: match[0]
            });
        }
        return results;
    }

    function decodeHtmlEntities(text) {
        if (!text) return '';
        return text
            .replace(/&lt;/g, '<')
            .replace(/&gt;/g, '>')
            .replace(/&amp;/g, '&')
            .replace(/&quot;/g, '"')
            .replace(/&apos;/g, "'")
            .replace(/&#(\d+);/g, (_, dec) => String.fromCharCode(dec))
            .replace(/&#x([0-9a-f]+);/gi, (_, hex) => String.fromCharCode(parseInt(hex, 16)));
    }

    function stripCdata(text) {
        if (!text) return '';
        return text.replace(/<!\[CDATA\[([\s\S]*?)\]\]>/g, '$1');
    }

    function cleanText(text) {
        if (!text) return '';
        return decodeHtmlEntities(stripCdata(text)).trim();
    }

    return {
        getTagContent,
        getAllTagContents,
        getTagAttribute,
        getAllTags,
        cleanText
    };
}

/**
 * Parse RSS 2.0 feed
 */
function parseRss(xml, parser) {
    const channel = parser.getTagContent(xml, 'channel');
    if (!channel) {
        throw new Error('Invalid RSS feed: no channel element found');
    }

    const feed = {
        title: parser.cleanText(parser.getTagContent(channel, 'title')) || '',
        link: parser.cleanText(parser.getTagContent(channel, 'link')) || '',
        description: parser.cleanText(parser.getTagContent(channel, 'description')) || '',
        items: []
    };

    const itemRegex = /<item[^>]*>([\s\S]*?)<\/item>/gi;
    let match;
    while ((match = itemRegex.exec(channel)) !== null) {
        const itemXml = match[1];

        const categories = parser.getAllTagContents(itemXml, 'category')
            .map(c => parser.cleanText(c));

        const item = {
            title: parser.cleanText(parser.getTagContent(itemXml, 'title')) || '',
            link: parser.cleanText(parser.getTagContent(itemXml, 'link')) || '',
            description: parser.cleanText(parser.getTagContent(itemXml, 'description')) || '',
            pubDate: parser.cleanText(parser.getTagContent(itemXml, 'pubDate')) || '',
            author: parser.cleanText(parser.getTagContent(itemXml, 'author') ||
                    parser.getTagContent(itemXml, 'dc:creator')) || '',
            categories: categories,
            guid: parser.cleanText(parser.getTagContent(itemXml, 'guid')) || ''
        };

        feed.items.push(item);
    }

    return feed;
}

/**
 * Parse Atom feed
 */
function parseAtom(xml, parser) {
    const feedContent = parser.getTagContent(xml, 'feed');
    if (!feedContent) {
        throw new Error('Invalid Atom feed: no feed element found');
    }

    function getAtomLink(content, rel) {
        const linkRegex = /<link([^>]*)>/gi;
        let match;
        while ((match = linkRegex.exec(content)) !== null) {
            const attrs = match[1];
            const relMatch = attrs.match(/rel=["']([^"']*)["']/);
            const hrefMatch = attrs.match(/href=["']([^"']*)["']/);

            if (hrefMatch) {
                const linkRel = relMatch ? relMatch[1] : 'alternate';
                if (linkRel === rel || (!rel && linkRel === 'alternate')) {
                    return hrefMatch[1];
                }
            }
        }
        return '';
    }

    const feed = {
        title: parser.cleanText(parser.getTagContent(feedContent, 'title')) || '',
        link: getAtomLink(feedContent, 'alternate') || getAtomLink(feedContent, null) || '',
        description: parser.cleanText(parser.getTagContent(feedContent, 'subtitle')) || '',
        items: []
    };

    const entryRegex = /<entry[^>]*>([\s\S]*?)<\/entry>/gi;
    let match;
    while ((match = entryRegex.exec(feedContent)) !== null) {
        const entryXml = match[1];

        const categories = [];
        const catRegex = /<category([^>]*)>/gi;
        let catMatch;
        while ((catMatch = catRegex.exec(entryXml)) !== null) {
            const termMatch = catMatch[1].match(/term=["']([^"']*)["']/);
            if (termMatch) {
                categories.push(parser.cleanText(termMatch[1]));
            }
        }

        let author = '';
        const authorContent = parser.getTagContent(entryXml, 'author');
        if (authorContent) {
            author = parser.cleanText(parser.getTagContent(authorContent, 'name')) || '';
        }

        const pubDate = parser.cleanText(
            parser.getTagContent(entryXml, 'published') ||
            parser.getTagContent(entryXml, 'updated')
        ) || '';

        const description = parser.cleanText(
            parser.getTagContent(entryXml, 'summary') ||
            parser.getTagContent(entryXml, 'content')
        ) || '';

        const item = {
            title: parser.cleanText(parser.getTagContent(entryXml, 'title')) || '',
            link: getAtomLink(entryXml, 'alternate') || getAtomLink(entryXml, null) || '',
            description: description,
            pubDate: pubDate,
            author: author,
            categories: categories,
            guid: parser.cleanText(parser.getTagContent(entryXml, 'id')) || ''
        };

        feed.items.push(item);
    }

    return feed;
}

/**
 * Parse date string to timestamp
 */
function parseDate(dateStr) {
    if (!dateStr) return 0;

    const timestamp = Date.parse(dateStr);
    if (!isNaN(timestamp)) {
        return timestamp;
    }

    const rfc822Regex = /(\d{1,2})\s+(\w{3})\s+(\d{4})\s+(\d{2}):(\d{2}):(\d{2})/;
    const match = dateStr.match(rfc822Regex);
    if (match) {
        const months = {
            Jan: 0, Feb: 1, Mar: 2, Apr: 3, May: 4, Jun: 5,
            Jul: 6, Aug: 7, Sep: 8, Oct: 9, Nov: 10, Dec: 11
        };
        const day = parseInt(match[1], 10);
        const month = months[match[2]];
        const year = parseInt(match[3], 10);
        const hour = parseInt(match[4], 10);
        const minute = parseInt(match[5], 10);
        const second = parseInt(match[6], 10);

        if (month !== undefined) {
            return new Date(year, month, day, hour, minute, second).getTime();
        }
    }

    return 0;
}

/**
 * Filter items by keyword
 */
function filterByKeyword(items, keyword) {
    if (!keyword) return items;

    const lowerKeyword = keyword.toLowerCase();
    return items.filter(item => {
        const title = (item.title || '').toLowerCase();
        const description = (item.description || '').toLowerCase();
        return title.includes(lowerKeyword) || description.includes(lowerKeyword);
    });
}

/**
 * Filter items by date
 */
function filterByDate(items, newerThanStr) {
    if (!newerThanStr) return items;

    const threshold = parseDate(newerThanStr);
    if (threshold === 0) {
        smartbotic.log.warn(`Could not parse date filter: ${newerThanStr}`);
        return items;
    }

    return items.filter(item => {
        const itemDate = parseDate(item.pubDate);
        return itemDate > threshold;
    });
}

/**
 * Limit number of items
 */
function limitItems(items, maxItems) {
    if (!maxItems || maxItems <= 0) return items;
    return items.slice(0, maxItems);
}

/**
 * Get unique identifier for an RSS/Atom item
 * Prefers GUID, falls back to link
 */
function getItemIdentifier(item) {
    return item.guid || item.link || '';
}

/**
 * Generate a stable document ID from feed URL for state storage
 */
function generateStateDocId(feedUrl) {
    // Create a simple hash from the feed URL for a stable doc ID
    let hash = 0;
    for (let i = 0; i < feedUrl.length; i++) {
        const char = feedUrl.charCodeAt(i);
        hash = ((hash << 5) - hash) + char;
        hash = hash & hash; // Convert to 32-bit integer
    }
    return 'rss-state-' + Math.abs(hash).toString(36);
}

/**
 * Load seen item identifiers from storage
 */
function loadSeenItems(collection, feedUrl) {
    const docId = generateStateDocId(feedUrl);

    try {
        const result = smartbotic.storage.get(collection, docId);

        if (result.found && result.document && result.document.seenIds) {
            smartbotic.log.debug(`Loaded ${result.document.seenIds.length} seen item IDs from storage`);
            return {
                seenIds: new Set(result.document.seenIds),
                docId: docId,
                exists: true,
                version: result.document._version || 1
            };
        }
    } catch (error) {
        smartbotic.log.warn(`Error loading seen items from storage: ${error.message}`);
    }

    return {
        seenIds: new Set(),
        docId: docId,
        exists: false,
        version: 0
    };
}

/**
 * Save seen item identifiers to storage
 */
function saveSeenItems(collection, docId, seenIds, feedUrl, exists, version) {
    const seenArray = Array.from(seenIds);
    const documentData = {
        feedUrl: feedUrl,
        seenIds: seenArray,
        lastUpdated: new Date().toISOString(),
        itemCount: seenArray.length
    };

    try {
        if (exists) {
            // Update existing document
            const result = smartbotic.storage.update(collection, docId, documentData, version, false);
            if (!result.success) {
                smartbotic.log.error(`Failed to update seen items: ${result.error}`);
                return false;
            }
            smartbotic.log.debug(`Updated ${seenArray.length} seen item IDs in storage`);
        } else {
            // Insert new document
            const result = smartbotic.storage.insert(collection, documentData, docId, 0);
            if (!result.success) {
                smartbotic.log.error(`Failed to save seen items: ${result.error}`);
                return false;
            }
            smartbotic.log.debug(`Saved ${seenArray.length} seen item IDs to storage`);
        }
        return true;
    } catch (error) {
        smartbotic.log.error(`Error saving seen items to storage: ${error.message}`);
        return false;
    }
}

/**
 * Filter items to only include new (unseen) items
 */
function filterNewItems(items, seenIds) {
    return items.filter(item => {
        const id = getItemIdentifier(item);
        return id && !seenIds.has(id);
    });
}

async function execute(config, input) {
    const url = input.url || config.url;
    const filterKeyword = input.filterKeyword || config.filterKeyword;
    const filterNewerThan = input.filterNewerThan || config.filterNewerThan;
    const maxItems = input.maxItems !== undefined ? input.maxItems : config.maxItems;
    const timeout = config.timeout || 30000;
    const detectNewItems = config.detectNewItems || false;
    const stateCollection = config.stateCollection;

    if (!url) {
        throw new Error('Feed URL is required');
    }

    // Validate stateful mode configuration
    if (detectNewItems && !stateCollection) {
        throw new Error('State collection is required when "Detect New Items" is enabled. Please select a read-write collection in the workflow storage settings.');
    }

    smartbotic.log.info(`Fetching RSS/Atom feed from ${url}`);

    const response = smartbotic.http.request({
        method: 'GET',
        url: url,
        timeout: timeout,
        headers: {
            'Accept': 'application/rss+xml, application/atom+xml, application/xml, text/xml, */*'
        }
    });

    if (response.status < 200 || response.status >= 300) {
        throw new Error(`Failed to fetch feed: HTTP ${response.status}`);
    }

    const xml = typeof response.data === 'string' ? response.data : JSON.stringify(response.data);

    if (!xml || xml.trim().length === 0) {
        throw new Error('Empty response from feed URL');
    }

    const parser = parseXml(xml);

    let feed;
    if (xml.includes('<feed') && xml.includes('xmlns') && xml.includes('atom')) {
        smartbotic.log.debug('Detected Atom feed format');
        feed = parseAtom(xml, parser);
    } else if (xml.includes('<rss') || xml.includes('<channel')) {
        smartbotic.log.debug('Detected RSS feed format');
        feed = parseRss(xml, parser);
    } else {
        throw new Error('Unknown feed format: neither RSS nor Atom detected');
    }

    let items = feed.items;
    const totalFeedItemCount = items.length;

    // Apply standard filters first (keyword, date)
    if (filterKeyword) {
        smartbotic.log.debug(`Filtering by keyword: ${filterKeyword}`);
        items = filterByKeyword(items, filterKeyword);
    }

    if (filterNewerThan) {
        smartbotic.log.debug(`Filtering items newer than: ${filterNewerThan}`);
        items = filterByDate(items, filterNewerThan);
    }

    // Stateful new item detection
    let newItemCount = items.length;
    let stateInfo = null;

    if (detectNewItems) {
        smartbotic.log.info(`Detecting new items using collection: ${stateCollection}`);

        // Load previously seen items
        stateInfo = loadSeenItems(stateCollection, url);

        // Filter to only new items
        const newItems = filterNewItems(items, stateInfo.seenIds);
        newItemCount = newItems.length;

        smartbotic.log.info(`Found ${newItemCount} new items out of ${items.length} filtered items`);

        // Update seen IDs with all current items (both new and previously seen)
        // This ensures we track all items from the current feed
        for (const item of items) {
            const id = getItemIdentifier(item);
            if (id) {
                stateInfo.seenIds.add(id);
            }
        }

        // Use only new items for output
        items = newItems;
    }

    // Apply max items limit last
    if (maxItems && maxItems > 0) {
        smartbotic.log.debug(`Limiting to ${maxItems} items`);
        items = limitItems(items, maxItems);
    }

    smartbotic.log.info(`Returning ${items.length} items from feed`);

    // Persist seen items state after successful processing
    if (detectNewItems && stateInfo) {
        const saved = saveSeenItems(
            stateCollection,
            stateInfo.docId,
            stateInfo.seenIds,
            url,
            stateInfo.exists,
            stateInfo.version
        );
        if (!saved) {
            smartbotic.log.warn('Failed to persist seen items state, but continuing with output');
        }
    }

    const result = {
        feedTitle: feed.title,
        feedLink: feed.link,
        feedDescription: feed.description,
        itemCount: items.length,
        items: items
    };

    // Add stateful mode info if enabled
    if (detectNewItems) {
        result.newItemCount = newItemCount;
        result.totalFeedItemCount = totalFeedItemCount;
    }

    return result;
}

module.exports = { configSchema, inputSchema, outputSchema, execute };