feat: US-006 - RSS Reader Node - Stateful New Item Detection

Add optional stateful tracking to detect which RSS items have been
previously seen. When enabled, only new (unseen) items are output.

- Add 'Detect new items' toggle (detectNewItems) to config schema
- Add collection selection (stateCollection) with dynamicOptions filtering
  for read-write collections configured on the workflow
- Store seen item GUIDs/links in user-selected collection
- Compare feed items against stored identifiers on each execution
- Output only new items when stateful mode enabled
- Persist seen identifiers after successful execution
- When disabled, maintain original stateless behavior (all items output)

nodes/rss/rss-reader.js

@@ -2,8 +2,8 @@
  * @node rss-reader
  * @name RSS Reader
  * @category rss
- * @version 1.0.0
- * @description Read and parse RSS/Atom feeds with optional filtering by keyword, date, or count
+ * @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
  */
 
@@ -35,6 +35,24 @@ const configSchema = {
             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']
@@ -69,6 +87,8 @@ const outputSchema = {
         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',
@@ -361,17 +381,123 @@ function limitItems(items, maxItems) {
     return items.slice(0, maxItems);
 }
 
-async function execute(config, input, context) {
+/**
+ * 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({
@@ -407,7 +533,9 @@ async function execute(config, input, context) {
     }
 
     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);
@@ -418,20 +546,73 @@ async function execute(config, input, context) {
         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(`Parsed ${items.length} items from feed`);
+    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');
+        }
+    }
 
-    return {
+    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 };