From fef7dac43fa0f4be59bf80f920c0cf06e827c711 Mon Sep 17 00:00:00 2001 From: Theodore Li Date: Fri, 11 Sep 2026 16:38:14 -0700 Subject: [PATCH] fix(rss): deliver unseen items published before the last poll --- apps/sim/lib/webhooks/polling/rss.test.ts | 128 ++++++++++++++++++++++ apps/sim/lib/webhooks/polling/rss.ts | 14 ++- 2 files changed, 136 insertions(+), 6 deletions(-) create mode 100644 apps/sim/lib/webhooks/polling/rss.test.ts diff --git a/apps/sim/lib/webhooks/polling/rss.test.ts b/apps/sim/lib/webhooks/polling/rss.test.ts new file mode 100644 index 00000000000..490e4a06bab --- /dev/null +++ b/apps/sim/lib/webhooks/polling/rss.test.ts @@ -0,0 +1,128 @@ +/** + * @vitest-environment node + */ +import { createLogger } from '@sim/logger' +import { createWorkflowRecord } from '@sim/testing' +import { beforeEach, describe, expect, it, vi } from 'vitest' + +const { mockFetch, mockValidateUrl, mockProcessEvent, mockUpdateConfig } = vi.hoisted(() => ({ + mockFetch: vi.fn(), + mockValidateUrl: vi.fn(), + mockProcessEvent: vi.fn(), + mockUpdateConfig: vi.fn(), +})) + +vi.mock('@/lib/core/security/input-validation.server', () => ({ + secureFetchWithPinnedIP: mockFetch, + validateUrlWithDNS: mockValidateUrl, +})) + +vi.mock('@/lib/core/idempotency/service', () => ({ + pollingIdempotency: { + executeWithIdempotency: vi.fn( + async (_provider: string, _key: string, execute: () => Promise) => execute() + ), + }, +})) + +vi.mock('@/lib/webhooks/processor', () => ({ + processPolledWebhookEvent: mockProcessEvent, +})) + +vi.mock('@/lib/webhooks/polling/utils', () => ({ + markWebhookSuccess: vi.fn(), + markWebhookFailed: vi.fn(), + updateWebhookProviderConfig: mockUpdateConfig, +})) + +import { rssPollingHandler } from '@/lib/webhooks/polling/rss' +import type { PollWebhookContext, WebhookRecord } from '@/lib/webhooks/polling/types' + +const SUBSCRIBED_AT = new Date('2026-08-27T18:36:16.000Z') +const LAST_CHECKED_AT = '2026-09-11T23:26:27.000Z' +const GUID = 'https://example.com/news/late-item' + +function context(lastSeenGuids: string[] = []): PollWebhookContext { + const webhookData: WebhookRecord = { + id: 'rss-webhook', + workflowId: 'rss-listener', + deploymentVersionId: null, + registrationStatus: null, + registrationGeneration: null, + configFingerprint: null, + preparedAt: null, + blockId: null, + path: 'rss-listener', + routingKey: null, + provider: 'rss', + providerConfig: { + feedUrl: 'https://example.com/feed.xml', + lastCheckedTimestamp: LAST_CHECKED_AT, + lastSeenGuids, + }, + isActive: true, + failedCount: 0, + lastFailedAt: null, + archivedAt: null, + createdAt: SUBSCRIBED_AT, + updatedAt: new Date(LAST_CHECKED_AT), + } + return { + webhookData, + workflowData: createWorkflowRecord({ + id: 'rss-listener', + }) as PollWebhookContext['workflowData'], + requestId: 'rss-request', + logger: createLogger('RssTest'), + } +} + +function feed(pubDate: string) { + return new Response( + ` + Canary fixturehttps://example.comRSS fixture + Late item${GUID}${pubDate} + `, + { headers: { 'Content-Type': 'application/rss+xml' } } + ) +} + +describe('RSS delivery across delayed feed updates', () => { + beforeEach(() => { + vi.clearAllMocks() + mockValidateUrl.mockResolvedValue({ isValid: true, resolvedIP: '203.0.113.1' }) + mockProcessEvent.mockResolvedValue({ success: true }) + mockUpdateConfig.mockResolvedValue(undefined) + }) + + it('delivers an unseen item published before the last poll but after subscription', async () => { + mockFetch.mockResolvedValue(feed('Fri, 11 Sep 2026 21:25:32 GMT')) + + expect(await rssPollingHandler.pollWebhook(context())).toBe('success') + expect(mockProcessEvent).toHaveBeenCalledExactlyOnceWith( + expect.anything(), + expect.anything(), + expect.objectContaining({ item: expect.objectContaining({ guid: GUID }) }), + 'rss-request' + ) + expect(mockUpdateConfig).toHaveBeenCalledWith( + 'rss-webhook', + expect.objectContaining({ lastSeenGuids: [GUID] }), + expect.anything() + ) + }) + + it('does not redeliver a known GUID when its publication date changes', async () => { + mockFetch.mockResolvedValue(feed('Fri, 11 Sep 2026 23:28:00 GMT')) + + expect(await rssPollingHandler.pollWebhook(context([GUID]))).toBe('success') + expect(mockProcessEvent).not.toHaveBeenCalled() + }) + + it('does not backfill items published before the subscription existed', async () => { + mockFetch.mockResolvedValue(feed('Thu, 27 Aug 2026 18:30:00 GMT')) + + expect(await rssPollingHandler.pollWebhook(context())).toBe('success') + expect(mockProcessEvent).not.toHaveBeenCalled() + }) +}) diff --git a/apps/sim/lib/webhooks/polling/rss.ts b/apps/sim/lib/webhooks/polling/rss.ts index 1fd4bb0affb..491d7756cd6 100644 --- a/apps/sim/lib/webhooks/polling/rss.ts +++ b/apps/sim/lib/webhooks/polling/rss.ts @@ -109,7 +109,7 @@ export const rssPollingHandler: PollingProviderHandler = { items: newItems, etag, lastModified, - } = await fetchNewRssItems(config, requestId, logger) + } = await fetchNewRssItems(config, webhookData.createdAt, requestId, logger) if (!newItems.length) { await updateRssState(webhookId, now.toISOString(), [], config, logger, etag, lastModified) @@ -195,6 +195,7 @@ async function updateRssState( async function fetchNewRssItems( config: RssWebhookConfig, + subscriptionStartedAt: Date, requestId: string, logger: Logger ): Promise<{ feed: RssFeed; items: RssItem[]; etag?: string; lastModified?: string }> { @@ -248,9 +249,6 @@ async function fetchNewRssItems( return { feed: feed as RssFeed, items: [], etag: newEtag, lastModified: newLastModified } } - const lastCheckedTime = config.lastCheckedTimestamp - ? new Date(config.lastCheckedTimestamp) - : null const lastSeenGuids = new Set(config.lastSeenGuids || []) const newItems = feed.items.filter((item) => { @@ -263,9 +261,13 @@ async function fetchNewRssItems( return false } - if (lastCheckedTime && item.isoDate) { + /** + * A cached feed can reveal an item after its publication time. Only the fixed + * subscription boundary excludes history; the last poll time is not a delivery cursor. + */ + if (item.isoDate) { const itemDate = new Date(item.isoDate) - if (itemDate <= lastCheckedTime) { + if (itemDate <= subscriptionStartedAt) { return false } }