Simplify polling, add manual refresh with sync context check

Co-authored-by: sam.willis <[email protected]>

Author: cursoragent

packages/rss-db-collection/README.md

@@ -279,7 +279,7 @@ const blogCollection = createCollection({
 })
 ```
 
-### Manual Polling Control
+### Manual Refresh
 
 ```typescript
 const collection = createCollection({
@@ -290,12 +290,9 @@ const collection = createCollection({
   })
 })
 
-// Control polling manually
-collection.utils.startPolling()
-console.log(`Polling: ${collection.utils.isPolling()}`) // true
-
-collection.utils.stopPolling()
-console.log(`Polling: ${collection.utils.isPolling()}`) // false
+// Manually refresh the feed
+await collection.utils.refresh()
+console.log('Feed refreshed!')
 
 // Get status
 console.log(`Seen items: ${collection.utils.getSeenItemsCount()}`)
@@ -381,10 +378,7 @@ Common error scenarios handled:
 ```typescript
 // Available on collection.utils for both RSS and Atom collections
 interface FeedCollectionUtils {
-  refresh(): Promise<void>           // Manual refresh (limited outside sync context)
-  startPolling(): void              // Start polling
-  stopPolling(): void               // Stop polling  
-  isPolling(): boolean              // Check polling status
+  refresh(): Promise<void>           // Manual feed refresh
   clearSeenItems(): void            // Clear deduplication cache
   getSeenItemsCount(): number       // Get number of tracked items
 }

packages/rss-db-collection/src/rss.ts

@@ -220,21 +220,6 @@ export interface FeedCollectionUtils extends UtilsRecord {
    */
   refresh: () => Promise<void>
 
-  /**
-   * Start polling if it was stopped
-   */
-  startPolling: () => void
-
-  /**
-   * Stop polling
-   */
-  stopPolling: () => void
-
-  /**
-   * Get the current polling status
-   */
-  isPolling: () => boolean
-
   /**
    * Clear the seen items cache
    */
@@ -475,8 +460,12 @@ function createFeedCollectionOptions<
 
   // State management
   let seenItems = new Map<string, { id: string; lastSeen: number }>()
-  let isPolling = false
   let pollingTimeoutId: NodeJS.Timeout | null = null
+  let syncParams:
+    | Parameters<
+        SyncConfig<ResolveType<TExplicit, TSchema, TFallback>, TKey>[`sync`]
+      >[0]
+    | null = null
 
   /**
    * Clean up old seen items to prevent memory leaks
@@ -515,7 +504,7 @@ function createFeedCollectionOptions<
   /**
    * Refresh feed data
    */
-  const refreshFeed = async (syncParams: {
+  const refreshFeed = async (params: {
     begin: () => void
     write: (message: {
       type: `insert` | `update` | `delete`
@@ -546,7 +535,7 @@ function createFeedCollectionOptions<
         throw new UnsupportedFeedFormatError(feedUrl)
       }
 
-      const { begin, write, commit } = syncParams
+      const { begin, write, commit } = params
       begin()
 
       let newItemsCount = 0
@@ -608,42 +597,10 @@ function createFeedCollectionOptions<
     }
   }
 
-  /**
-   * Start polling
-   */
-  const startPollingFn = (syncParams?: any) => {
-    if (isPolling) {
-      return // Already polling
-    }
-
-    isPolling = true
-
-    const poll = async () => {
-      if (!isPolling) {
-        return // Polling was stopped
-      }
-
-      try {
-        if (syncParams) {
-          await refreshFeed(syncParams)
-        }
-      } catch (error) {
-        debug(`Polling error: ${error}`)
-        // Continue polling despite errors
-      }
-
-      // Schedule next poll
-      pollingTimeoutId = setTimeout(poll, pollingInterval)
-    }
-
-    poll()
-  }
-
   /**
    * Stop polling
    */
-  const stopPollingFn = () => {
-    isPolling = false
+  const stopPolling = () => {
     if (pollingTimeoutId) {
       clearTimeout(pollingTimeoutId)
       pollingTimeoutId = null
@@ -657,14 +614,32 @@ function createFeedCollectionOptions<
     sync: (params) => {
       const { markReady } = params
 
+      // Store sync params for manual refresh
+      syncParams = params
+
+      // Polling function
+      const poll = async () => {
+        try {
+          await refreshFeed(syncParams!)
+        } catch (error) {
+          debug(`Polling error: ${error}`)
+          // Continue polling despite errors
+        }
+
+        // Schedule next poll if polling is enabled
+        if (startPolling) {
+          pollingTimeoutId = setTimeout(poll, pollingInterval)
+        }
+      }
+
       // Initial feed fetch
       refreshFeed(params)
         .then(() => {
           markReady()
 
           // Start polling if configured to do so
           if (startPolling) {
-            startPollingFn(params)
+            pollingTimeoutId = setTimeout(poll, pollingInterval)
           }
         })
         .catch((error) => {
@@ -673,29 +648,26 @@ function createFeedCollectionOptions<
 
           // Still start polling for retry attempts
           if (startPolling) {
-            startPollingFn(params)
+            pollingTimeoutId = setTimeout(poll, pollingInterval)
           }
         })
 
       // Return cleanup function
       return () => {
-        stopPollingFn()
+        stopPolling()
+        syncParams = null
       }
     },
   }
 
   // Utils
   const utils: FeedCollectionUtils = {
-    refresh: () => {
-      // For manual refresh, we need access to sync params
-      // This is a limitation - manual refresh without sync params
-      return Promise.reject(
-        new Error(`Manual refresh not supported outside of sync context`)
-      )
+    refresh: async () => {
+      if (!syncParams) {
+        throw new Error(`Collection not synced yet - cannot refresh`)
+      }
+      await refreshFeed(syncParams)
     },
-    startPolling: () => startPollingFn(),
-    stopPolling: stopPollingFn,
-    isPolling: () => isPolling,
     clearSeenItems: () => {
       seenItems = new Map()
     },

packages/rss-db-collection/tests/rss.test.ts

@@ -261,7 +261,7 @@ describe(`RSS Collection`, () => {
       expect(fetchMock).toHaveBeenCalledTimes(3)
     })
 
-    it(`should allow manual polling control`, async () => {
+    it(`should allow manual refresh`, async () => {
       const fetchMock = vi.fn().mockResolvedValue({
         ok: true,
         text: () => Promise.resolve(sampleRSSFeed),
@@ -284,25 +284,32 @@ describe(`RSS Collection`, () => {
 
       expect(fetchMock).toHaveBeenCalledTimes(1) // Initial fetch only
 
-      // Start polling manually
-      collection.utils.startPolling()
-      expect(collection.utils.isPolling()).toBe(true)
-
-      // Advance time
-      vi.advanceTimersByTime(10000)
-      await flushPromises()
+      // Manually refresh the feed
+      await collection.utils.refresh()
 
       expect(fetchMock).toHaveBeenCalledTimes(2)
 
-      // Stop polling
-      collection.utils.stopPolling()
-      expect(collection.utils.isPolling()).toBe(false)
+      // Refresh again
+      await collection.utils.refresh()
 
-      // Advance time - should not fetch again
-      vi.advanceTimersByTime(10000)
-      await flushPromises()
+      expect(fetchMock).toHaveBeenCalledTimes(3)
+    })
+
+    it(`should throw error when refresh is called before sync`, async () => {
+      const config: RSSCollectionConfig = {
+        feedUrl: `https://example.com/rss.xml`,
+        pollingInterval: 10000,
+        getKey: (item: any) => item.guid || item.link,
+        startPolling: false,
+      }
 
-      expect(fetchMock).toHaveBeenCalledTimes(2) // No additional fetch
+      const options = rssCollectionOptions(config)
+      const collection = createCollection(options)
+
+      // Try to refresh before collection has synced
+      await expect(collection.utils.refresh()).rejects.toThrow(
+        `Collection not synced yet - cannot refresh`
+      )
     })
   })