smartbotic/nodes/rss/rss-reader.js

/**
 * @node rss-reader
 * @name RSS Reader
 * @category data
 * @version 1.2.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
        },
        userAgent: {
            type: 'string',
            title: 'User-Agent',
            description: 'Sent with the request. Several feeds refuse an unnamed client outright - Reddit answers 403 - and some rate-limit a browser string harder than a named one. Identify yourself here',
            default: 'smartbotic-rss/1.0 (+https://smartbotics.ai)'
        },
        timeout: {
            type: 'number',
            title: 'Timeout (ms)',
            default: 30000
        },
        retries: {
            type: 'number', title: 'Retries', default: 2,
            description: 'Extra attempts when the feed times out or answers 429, 500, 502, 503 or 504. A feed that is briefly unreachable should not end a scheduled run'
        },
        retryDelayMs: {
            type: 'number', title: 'Retry Delay (ms)', default: 2000,
            description: 'Wait before the first retry. Doubles on each further attempt, and a Retry-After from the server wins over both'
        },
        retryMaxDelayMs: {
            type: 'number', title: 'Max Retry Delay (ms)', default: 30000,
            description: 'Upper bound for the backoff'
        },
        skipOnError: {
            type: 'boolean', title: 'Skip On Error', default: false,
            description: 'Return success false and no items instead of failing the workflow. For a workflow reading several feeds, this is what lets the others carry on when one is unreachable - without it the first bad feed ends the run'
        },
        skipTlsVerify: {
            type: 'boolean', title: 'Skip TLS certificate verification', default: false,
            description: 'Accept the feed even if its certificate has expired or does not chain to a ' +
                         'trusted root. The hostname is still checked, so this cannot be tricked into ' +
                         'accepting a certificate for a different site - only a broken one for the right ' +
                         'site. An attacker who can intercept this connection can impersonate the feed ' +
                         'host while this is on. Use it only for a feed you control, or a certificate ' +
                         'failure you have personally checked - not as a standing setting.'
        },
        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: {
        success: { type: 'boolean', description: 'False when the feed could not be read and Skip On Error let the run carry on' },
        error: { type: 'string', description: 'Why it could not be read, when it could not' },
        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: 'Items this run will process. With Detect New Items off that is every item returned, so a workflow can branch on this whichever mode it is in' },
        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', description: 'Item title' },
                    link: { type: 'string', description: 'Item link/URL' },
                    description: { type: 'string', description: 'Item summary/description' },
                    content: { type: 'string', description: 'Full content (content:encoded)' },
                    pubDate: { type: 'string', description: 'Publication date' },
                    author: { type: 'string', description: 'Author name' },
                    categories: { type: 'array', items: { type: 'string' }, description: 'Categories/tags' },
                    guid: { type: 'string', description: 'Unique identifier' },
                    comments: { type: 'string', description: 'Comments URL' },
                    source: { type: 'string', description: 'Source attribution' },
                    enclosure: {
                        type: 'object',
                        description: 'Media enclosure (attachment)',
                        properties: {
                            url: { type: 'string', description: 'Media URL' },
                            type: { type: 'string', description: 'MIME type' },
                            length: { type: 'number', description: 'File size in bytes' }
                        }
                    },
                    media: {
                        type: 'object',
                        description: 'Media content (media:content)',
                        properties: {
                            url: { type: 'string' },
                            type: { type: 'string' },
                            width: { type: 'number' },
                            height: { type: 'number' }
                        }
                    },
                    thumbnail: {
                        type: 'object',
                        description: 'Thumbnail image (media:thumbnail)',
                        properties: {
                            url: { type: 'string' },
                            width: { type: 'number' },
                            height: { type: 'number' }
                        }
                    }
                }
            }
        }
    }
};

/**
 * Simple XML parser utilities for RSS/Atom feeds
 * Handles basic XML structure without external dependencies
 */
function xmlGetTagContent(xml, tagName) {
    var pattern = new RegExp('<' + tagName + '[^>]*>([\\s\\S]*?)<\\/' + tagName + '>', 'i');
    var match = xml.match(pattern);
    if (match) {
        return match[1].trim();
    }
    return null;
}

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

function xmlGetTagAttributes(xml, tagName) {
    // Match self-closing tag or opening tag
    var regex = new RegExp('<' + tagName + '([^>]*?)\\/?>', 'i');
    var match = xml.match(regex);
    if (!match) return null;

    var attrString = match[1];
    var attrs = {};

    // Parse attributes: name="value" or name='value'
    var attrRegex = /(\w+)=["']([^"']*)["']/g;
    var attrMatch;
    while ((attrMatch = attrRegex.exec(attrString)) !== null) {
        attrs[attrMatch[1]] = attrMatch[2];
    }

    return attrs;
}

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

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

function xmlCleanText(text) {
    if (!text) return '';
    return xmlDecodeHtmlEntities(xmlStripCdata(text)).trim();
}

/**
 * Parse RSS 2.0 feed
 */
function parseRss(xml) {
    var channel = xmlGetTagContent(xml, 'channel');
    if (!channel) {
        throw new Error('Invalid RSS feed: no channel element found');
    }
    var feedTitle = xmlGetTagContent(channel, 'title');
    var feedLink = xmlGetTagContent(channel, 'link');
    var feedDesc = xmlGetTagContent(channel, 'description');
    var feedLanguage = xmlGetTagContent(channel, 'language');
    var feedPubDate = xmlGetTagContent(channel, 'pubDate');
    var feedLastBuildDate = xmlGetTagContent(channel, 'lastBuildDate');
    var feedGenerator = xmlGetTagContent(channel, 'generator');
    var feedImage = xmlGetTagContent(channel, 'image');

    var feed = {
        title: xmlCleanText(feedTitle) || '',
        link: xmlCleanText(feedLink) || '',
        description: xmlCleanText(feedDesc) || '',
        language: xmlCleanText(feedLanguage) || '',
        pubDate: xmlCleanText(feedPubDate) || '',
        lastBuildDate: xmlCleanText(feedLastBuildDate) || '',
        generator: xmlCleanText(feedGenerator) || '',
        items: []
    };

    // Parse feed image if present
    if (feedImage) {
        feed.image = {
            url: xmlCleanText(xmlGetTagContent(feedImage, 'url')) || '',
            title: xmlCleanText(xmlGetTagContent(feedImage, 'title')) || '',
            link: xmlCleanText(xmlGetTagContent(feedImage, 'link')) || ''
        };
    }

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

        var catArray = xmlGetAllTagContents(itemXml, 'category');
        var categories = [];
        for (var ci = 0; ci < catArray.length; ci++) {
            categories.push(xmlCleanText(catArray[ci]));
        }

        var item = {
            title: xmlCleanText(xmlGetTagContent(itemXml, 'title')) || '',
            link: xmlCleanText(xmlGetTagContent(itemXml, 'link')) || '',
            description: xmlCleanText(xmlGetTagContent(itemXml, 'description')) || '',
            content: xmlCleanText(xmlGetTagContent(itemXml, 'content:encoded')) || '',
            pubDate: xmlCleanText(xmlGetTagContent(itemXml, 'pubDate')) || '',
            author: xmlCleanText(xmlGetTagContent(itemXml, 'author') ||
                    xmlGetTagContent(itemXml, 'dc:creator')) || '',
            categories: categories,
            guid: xmlCleanText(xmlGetTagContent(itemXml, 'guid')) || '',
            comments: xmlCleanText(xmlGetTagContent(itemXml, 'comments')) || '',
            source: xmlCleanText(xmlGetTagContent(itemXml, 'source')) || ''
        };

        // Parse enclosure (media attachment)
        var enclosureAttrs = xmlGetTagAttributes(itemXml, 'enclosure');
        if (enclosureAttrs) {
            item.enclosure = {
                url: enclosureAttrs.url || '',
                type: enclosureAttrs.type || '',
                length: enclosureAttrs.length ? parseInt(enclosureAttrs.length, 10) : 0
            };
        }

        // Parse media:content (common in media RSS)
        var mediaAttrs = xmlGetTagAttributes(itemXml, 'media:content');
        if (mediaAttrs) {
            item.media = {
                url: mediaAttrs.url || '',
                type: mediaAttrs.type || mediaAttrs.medium || '',
                width: mediaAttrs.width ? parseInt(mediaAttrs.width, 10) : 0,
                height: mediaAttrs.height ? parseInt(mediaAttrs.height, 10) : 0
            };
        }

        // Parse media:thumbnail
        var thumbAttrs = xmlGetTagAttributes(itemXml, 'media:thumbnail');
        if (thumbAttrs) {
            item.thumbnail = {
                url: thumbAttrs.url || '',
                width: thumbAttrs.width ? parseInt(thumbAttrs.width, 10) : 0,
                height: thumbAttrs.height ? parseInt(thumbAttrs.height, 10) : 0
            };
        }

        feed.items.push(item);
    }

    return feed;
}

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

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

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

    function getAtomLinkByType(content, type) {
        var linkRegex = /<link([^>]*)>/gi;
        var match;
        while ((match = linkRegex.exec(content)) !== null) {
            var attrs = match[1];
            var typeMatch = attrs.match(/type=["']([^"']*)["']/);
            var hrefMatch = attrs.match(/href=["']([^"']*)["']/);

            if (hrefMatch && typeMatch && typeMatch[1].indexOf(type) !== -1) {
                return {
                    url: hrefMatch[1],
                    type: typeMatch[1]
                };
            }
        }
        return null;
    }

    var feed = {
        title: xmlCleanText(xmlGetTagContent(feedContent, 'title')) || '',
        link: getAtomLink(feedContent, 'alternate') || getAtomLink(feedContent, null) || '',
        description: xmlCleanText(xmlGetTagContent(feedContent, 'subtitle')) || '',
        language: '',
        pubDate: xmlCleanText(xmlGetTagContent(feedContent, 'updated')) || '',
        lastBuildDate: '',
        generator: xmlCleanText(xmlGetTagContent(feedContent, 'generator')) || '',
        items: []
    };

    // Parse feed icon/logo
    var feedIcon = xmlGetTagContent(feedContent, 'icon');
    var feedLogo = xmlGetTagContent(feedContent, 'logo');
    if (feedIcon || feedLogo) {
        feed.image = {
            url: xmlCleanText(feedLogo || feedIcon) || '',
            title: feed.title,
            link: feed.link
        };
    }

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

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

        var author = '';
        var authorContent = xmlGetTagContent(entryXml, 'author');
        if (authorContent) {
            author = xmlCleanText(xmlGetTagContent(authorContent, 'name')) || '';
        }

        var pubDate = xmlCleanText(
            xmlGetTagContent(entryXml, 'published') ||
            xmlGetTagContent(entryXml, 'updated')
        ) || '';

        var summary = xmlCleanText(xmlGetTagContent(entryXml, 'summary')) || '';
        var content = xmlCleanText(xmlGetTagContent(entryXml, 'content')) || '';

        var item = {
            title: xmlCleanText(xmlGetTagContent(entryXml, 'title')) || '',
            link: getAtomLink(entryXml, 'alternate') || getAtomLink(entryXml, null) || '',
            description: summary || content,
            content: content,
            pubDate: pubDate,
            author: author,
            categories: categories,
            guid: xmlCleanText(xmlGetTagContent(entryXml, 'id')) || '',
            comments: '',
            source: ''
        };

        // Check for enclosure link
        var enclosureLink = getAtomLinkByType(entryXml, 'image');
        if (!enclosureLink) {
            enclosureLink = getAtomLinkByType(entryXml, 'audio');
        }
        if (!enclosureLink) {
            enclosureLink = getAtomLinkByType(entryXml, 'video');
        }
        if (enclosureLink) {
            item.enclosure = {
                url: enclosureLink.url,
                type: enclosureLink.type,
                length: 0
            };
        }

        // Parse media:content
        var mediaAttrs = xmlGetTagAttributes(entryXml, 'media:content');
        if (mediaAttrs) {
            item.media = {
                url: mediaAttrs.url || '',
                type: mediaAttrs.type || mediaAttrs.medium || '',
                width: mediaAttrs.width ? parseInt(mediaAttrs.width, 10) : 0,
                height: mediaAttrs.height ? parseInt(mediaAttrs.height, 10) : 0
            };
        }

        // Parse media:thumbnail
        var thumbAttrs = xmlGetTagAttributes(entryXml, 'media:thumbnail');
        if (thumbAttrs) {
            item.thumbnail = {
                url: thumbAttrs.url || '',
                width: thumbAttrs.width ? parseInt(thumbAttrs.width, 10) : 0,
                height: thumbAttrs.height ? parseInt(thumbAttrs.height, 10) : 0
            };
        }

        feed.items.push(item);
    }

    return feed;
}

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

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

    var rfc822Regex = /(\d{1,2})\s+(\w{3})\s+(\d{4})\s+(\d{2}):(\d{2}):(\d{2})/;
    var match = dateStr.match(rfc822Regex);
    if (match) {
        var 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
        };
        var day = parseInt(match[1], 10);
        var month = months[match[2]];
        var year = parseInt(match[3], 10);
        var hour = parseInt(match[4], 10);
        var minute = parseInt(match[5], 10);
        var 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;

    var lowerKeyword = keyword.toLowerCase();
    var result = [];
    for (var i = 0; i < items.length; i++) {
        var item = items[i];
        var title = (item.title || '').toLowerCase();
        var description = (item.description || '').toLowerCase();
        if (title.indexOf(lowerKeyword) !== -1 || description.indexOf(lowerKeyword) !== -1) {
            result.push(item);
        }
    }
    return result;
}

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

    var threshold = parseDate(newerThanStr);
    if (threshold === 0) {
        smartbotic.log.warn('Could not parse date filter: ' + newerThanStr);
        return items;
    }

    var result = [];
    for (var i = 0; i < items.length; i++) {
        var item = items[i];
        var itemDate = parseDate(item.pubDate);
        if (itemDate > threshold) {
            result.push(item);
        }
    }
    return result;
}

/**
 * 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
    var hash = 0;
    for (var i = 0; i < feedUrl.length; i++) {
        var 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) {
    var docId = generateStateDocId(feedUrl);

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

        if (result.found && result.document && result.document.seenIds) {
            // Use plain object as a set (for QuickJS compatibility)
            var seenObj = {};
            var storedIds = result.document.seenIds;
            // Handle both array format and object format (in case of legacy data)
            if (Array.isArray(storedIds)) {
                smartbotic.log.debug('Loaded ' + storedIds.length + ' seen item IDs from storage');
                for (var i = 0; i < storedIds.length; i++) {
                    seenObj[storedIds[i]] = true;
                }
            } else if (typeof storedIds === 'object' && storedIds !== null) {
                // Object format - keys are the IDs
                var keys = Object.keys(storedIds);
                smartbotic.log.debug('Loaded ' + keys.length + ' seen item IDs from storage (object format)');
                for (var j = 0; j < keys.length; j++) {
                    seenObj[keys[j]] = true;
                }
            }
            return {
                seenIds: seenObj,
                docId: docId,
                exists: true,
                version: result.document._version || 1
            };
        }
    } catch (error) {
        smartbotic.log.warn('Error loading seen items from storage: ' + error.message);
    }

    return {
        seenIds: {},
        docId: docId,
        exists: false,
        version: 0
    };
}

/**
 * Save seen item identifiers to storage
 */
function saveSeenItems(collection, docId, seenIds, feedUrl, exists, version) {
    // Convert object keys to array
    var seenArray = Object.keys(seenIds);
    var documentData = {
        feedUrl: feedUrl,
        seenIds: seenArray,
        lastUpdated: new Date().toISOString(),
        itemCount: seenArray.length
    };

    try {
        if (exists) {
            // Update existing document
            var updateResult = smartbotic.storage.update(collection, docId, documentData, version, false);
            if (!updateResult.success) {
                smartbotic.log.error('Failed to update seen items: ' + updateResult.error);
                return false;
            }
            smartbotic.log.debug('Updated ' + seenArray.length + ' seen item IDs in storage');
        } else {
            // Insert new document
            var insertResult = smartbotic.storage.insert(collection, documentData, docId, 0);
            if (!insertResult.success) {
                // If insert fails because document exists, try update instead
                if (insertResult.error && insertResult.error.indexOf('already exists') !== -1) {
                    smartbotic.log.debug('Document exists, trying update instead');
                    var retryResult = smartbotic.storage.update(collection, docId, documentData, 0, false);
                    if (!retryResult.success) {
                        smartbotic.log.error('Failed to update seen items on retry: ' + retryResult.error);
                        return false;
                    }
                    smartbotic.log.debug('Updated ' + seenArray.length + ' seen item IDs in storage (retry)');
                } else {
                    smartbotic.log.error('Failed to save seen items: ' + insertResult.error);
                    return false;
                }
            } else {
                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) {
    var newItems = [];
    for (var i = 0; i < items.length; i++) {
        var id = getItemIdentifier(items[i]);
        if (id && !seenIds[id]) {
            newItems.push(items[i]);
        }
    }
    return newItems;
}

async function readFeed(config, input) {
    var url = input.url || config.url;
    var filterKeyword = input.filterKeyword || config.filterKeyword;
    var filterNewerThan = input.filterNewerThan || config.filterNewerThan;
    var maxItems = input.maxItems !== undefined ? input.maxItems : config.maxItems;
    var timeout = config.timeout || 30000;
    var detectNewItems = config.detectNewItems || false;
    var 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.');
    }

    // No User-Agent at all is what several feeds reject, Reddit among them, so
    // there is always one.
    var userAgent = String(config.userAgent || '').trim() ||
                    'smartbotic-rss/1.0 (+https://smartbotics.ai)';

    smartbotic.log.info('Fetching RSS/Atom feed from ' + url);

    var response = smartbotic.http.request({
        method: 'GET',
        url: url,
        timeout: timeout,
        // Reading a feed is safe to repeat, and a feed that times out once is
        // the usual reason a scheduled run dies - the request is retried rather
        // than the whole pass being thrown away. Handled by the HTTP helper, so
        // the waiting happens without this node holding a loop.
        retries: config.retries !== undefined && config.retries !== null && config.retries !== ''
            ? Number(config.retries) : 2,
        retryDelayMs: Number(config.retryDelayMs) || 2000,
        retryMaxDelayMs: Number(config.retryMaxDelayMs) || 30000,
        skipTlsVerify: config.skipTlsVerify === true,
        headers: {
            'Accept': 'application/rss+xml, application/atom+xml, application/xml, text/xml, */*',
            'User-Agent': userAgent
        }
    });

    if (response.status < 200 || response.status >= 300) {
        var hint = '';
        if (response.status === 403) {
            hint = '. The feed refused this client - most feeds that do this want a descriptive ' +
                   'User-Agent, which is the User-Agent setting on this node';
        } else if (response.status === 429) {
            hint = '. The feed is rate limiting this address - poll it less often, or from ' +
                   'somewhere else. Reddit throttles hard and counts every request, including ' +
                   'failed ones';
        } else if (response.status === 404) {
            hint = '. Check the address: ' + url;
        }
        throw new Error('Failed to fetch feed: HTTP ' + response.status + hint);
    }

    var 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');
    }

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

    var items = feed.items;
    var 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
    var newItemCount = items.length;
    var 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
        var 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 (var i = 0; i < items.length; i++) {
            var id = getItemIdentifier(items[i]);
            if (id) {
                stateInfo.seenIds[id] = true;
            }
        }

        // 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) {
        var 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');
        }
    }

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

    // Always reported, in both modes.
    //
    // These used to appear only when Detect New Items was on, so a workflow
    // branching on newItemCount > 0 quietly started taking the other branch the
    // moment someone turned detection off: the field was not zero, it was gone,
    // and a comparison against a missing field is false. With detection off
    // every item returned is one this run will process, so newItemCount is the
    // item count - which is what a workflow asking "is there anything to do?"
    // means either way.
    result.newItemCount = detectNewItems ? newItemCount : items.length;
    result.totalFeedItemCount = detectNewItems ? totalFeedItemCount : items.length;

    return result;
}

async function execute(config, input) {
    try {
        var result = await readFeed(config, input);
        result.success = true;
        result.error = '';
        return result;
    } catch (err) {
        if (config.skipOnError !== true) {
            throw err;
        }
        var message = (err && err.message) ? err.message : String(err);
        smartbotic.log.warn('RSS Reader: ' + (config.url || '') + ' could not be read, skipping: ' + message);
        // Shaped like a successful read with nothing in it, so a merge or a
        // loop downstream does not have to know the difference - and carrying
        // the reason means a run that produced nothing can still say why.
        return {
            success: false,
            error: message,
            feedTitle: '',
            feedLink: '',
            feedDescription: '',
            itemCount: 0,
            items: [],
            newItemCount: 0,
            totalFeedItemCount: 0
        };
    }
}

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