From 7b9f5451290017478be9eb10bdf52532f1fb3d98 Mon Sep 17 00:00:00 2001 From: Zita Szupera Date: Fri, 18 Sep 2026 08:10:24 -0500 Subject: [PATCH] fix: don't rewatch a feed on reconnect after stopWatching was called --- .../feed-watch-unwatch.test.ts | 27 ++++ .../watch/handle-watch-started.test.ts | 54 +++++++ .../watch/handle-watch-started.ts | 30 +++- .../watch/handle-watch-stopped.test.ts | 78 ++++++++++ .../watch/handle-watch-stopped.ts | 30 +++- packages/feeds-client/src/feed/feed.test.ts | 15 ++ .../src/feeds-client/feeds-client.test.ts | 139 ++++++++++++++++++ .../src/feeds-client/feeds-client.ts | 10 +- 8 files changed, 376 insertions(+), 7 deletions(-) create mode 100644 packages/feeds-client/src/feed/event-handlers/watch/handle-watch-started.test.ts create mode 100644 packages/feeds-client/src/feed/event-handlers/watch/handle-watch-stopped.test.ts diff --git a/packages/feeds-client/__integration-tests__/feed-watch-unwatch.test.ts b/packages/feeds-client/__integration-tests__/feed-watch-unwatch.test.ts index 98b8a2c2..6b497cb0 100644 --- a/packages/feeds-client/__integration-tests__/feed-watch-unwatch.test.ts +++ b/packages/feeds-client/__integration-tests__/feed-watch-unwatch.test.ts @@ -105,6 +105,33 @@ describe('Feed watch and unwatch', () => { expect(activity['feed']?.currentState.watch).toBe(true); }); + it(`stop watching prevents the feed from being refetched on reconnect`, async () => { + // a dedicated feed: the reconnect test above leaves an unawaited + // getOrCreate in flight on the shared one + const ownFeed = client.feed(feedGroup, crypto.randomUUID()); + await ownFeed.getOrCreate({ watch: true }); + expect(ownFeed.currentState.watch).toBe(true); + + await ownFeed.stopWatching(); + expect(ownFeed.currentState.watch).toBe(false); + + const spy = vi.spyOn(ownFeed, 'getOrCreate'); + + client['eventDispatcher'].dispatch({ + type: 'connection.changed', + online: false, + }); + client['eventDispatcher'].dispatch({ + type: 'connection.changed', + online: true, + }); + + expect(spy).not.toHaveBeenCalled(); + + spy.mockRestore(); + await ownFeed.delete(); + }); + afterAll(async () => { await feed.delete(); await client.disconnectUser(); diff --git a/packages/feeds-client/src/feed/event-handlers/watch/handle-watch-started.test.ts b/packages/feeds-client/src/feed/event-handlers/watch/handle-watch-started.test.ts new file mode 100644 index 00000000..7d4a8b7b --- /dev/null +++ b/packages/feeds-client/src/feed/event-handlers/watch/handle-watch-started.test.ts @@ -0,0 +1,54 @@ +import { beforeEach, describe, expect, it } from 'vitest'; +import { FeedsClient } from '../../../feeds-client'; +import { Feed } from '../../feed'; +import { generateFeedResponse } from '../../../test-utils'; +import { handleWatchStarted } from './handle-watch-started'; + +describe('handleWatchStarted', () => { + let feed: Feed; + + beforeEach(() => { + const client = new FeedsClient('mock-api-key'); + const feedResponse = generateFeedResponse({ id: 'main', group_id: 'user' }); + feed = new Feed( + client, + feedResponse.group_id, + feedResponse.id, + feedResponse, + ); + }); + + it('sets watch and leaves the stored config untouched by default', () => { + const config = { watch: false, limit: 20 }; + feed.state.partialNext({ last_get_or_create_request_config: config }); + + handleWatchStarted.call(feed); + + expect(feed.currentState.watch).toBe(true); + expect(feed.currentState.last_get_or_create_request_config).toBe(config); + }); + + it('restores the watch intent with setWatchIntent', () => { + const config = { watch: false, limit: 20 }; + feed.state.partialNext({ last_get_or_create_request_config: config }); + + handleWatchStarted.call(feed, { setWatchIntent: true }); + + expect(feed.currentState.watch).toBe(true); + expect(feed.currentState.last_get_or_create_request_config).toEqual({ + watch: true, + limit: 20, + }); + expect(feed.currentState.last_get_or_create_request_config).not.toBe( + config, + ); + expect(config.watch).toBe(false); + }); + + it('does not invent a config when none was stored', () => { + handleWatchStarted.call(feed, { setWatchIntent: true }); + + expect(feed.currentState.watch).toBe(true); + expect(feed.currentState.last_get_or_create_request_config).toBeUndefined(); + }); +}); diff --git a/packages/feeds-client/src/feed/event-handlers/watch/handle-watch-started.ts b/packages/feeds-client/src/feed/event-handlers/watch/handle-watch-started.ts index 6320cd06..0957460c 100644 --- a/packages/feeds-client/src/feed/event-handlers/watch/handle-watch-started.ts +++ b/packages/feeds-client/src/feed/event-handlers/watch/handle-watch-started.ts @@ -1,5 +1,31 @@ import type { Feed } from '../../feed'; -export function handleWatchStarted(this: Feed) { - this.state.partialNext({ watch: true }); +export function handleWatchStarted( + this: Feed, + { + setWatchIntent = false, + }: { + /** + * Also restores the watch intent in `last_get_or_create_request_config`, so a + * feed that was explicitly stop-watched and is now watched again is re-fetched + * by `synchronize()` on the next reconnect. Only meaningful when a config is + * already stored — feeds watched solely through `queryFeeds` have none. + */ + setWatchIntent?: boolean; + } = {}, +) { + if (!setWatchIntent) { + this.state.partialNext({ watch: true }); + return; + } + + this.state.next((currentState) => ({ + ...currentState, + watch: true, + last_get_or_create_request_config: + currentState.last_get_or_create_request_config && { + ...currentState.last_get_or_create_request_config, + watch: true, + }, + })); } diff --git a/packages/feeds-client/src/feed/event-handlers/watch/handle-watch-stopped.test.ts b/packages/feeds-client/src/feed/event-handlers/watch/handle-watch-stopped.test.ts new file mode 100644 index 00000000..9e5f11ff --- /dev/null +++ b/packages/feeds-client/src/feed/event-handlers/watch/handle-watch-stopped.test.ts @@ -0,0 +1,78 @@ +import { beforeEach, describe, expect, it } from 'vitest'; +import { FeedsClient } from '../../../feeds-client'; +import { Feed } from '../../feed'; +import { generateFeedResponse } from '../../../test-utils'; +import { handleWatchStopped } from './handle-watch-stopped'; + +describe('handleWatchStopped', () => { + let feed: Feed; + + beforeEach(() => { + const client = new FeedsClient('mock-api-key'); + const feedResponse = generateFeedResponse({ id: 'main', group_id: 'user' }); + feed = new Feed( + client, + feedResponse.group_id, + feedResponse.id, + feedResponse, + true, + ); + }); + + describe('without clearWatchIntent (connection lost)', () => { + it('clears watch but keeps the watch intent so reconnect can recover', () => { + const config = { watch: true, limit: 20 }; + feed.state.partialNext({ last_get_or_create_request_config: config }); + + handleWatchStopped.call(feed); + + expect(feed.currentState.watch).toBe(false); + // same reference - the stored config is left completely untouched + expect(feed.currentState.last_get_or_create_request_config).toBe(config); + }); + }); + + describe('with clearWatchIntent (explicit stopWatching)', () => { + it('clears watch and the stored watch intent', () => { + feed.state.partialNext({ + last_get_or_create_request_config: { watch: true, limit: 20 }, + }); + + handleWatchStopped.call(feed, { clearWatchIntent: true }); + + expect(feed.currentState.watch).toBe(false); + expect(feed.currentState.last_get_or_create_request_config).toEqual({ + watch: false, + limit: 20, + }); + }); + + it('produces a new last_get_or_create_request_config reference', () => { + const config = { watch: true, limit: 20 }; + feed.state.partialNext({ last_get_or_create_request_config: config }); + + handleWatchStopped.call(feed, { clearWatchIntent: true }); + + expect(feed.currentState.last_get_or_create_request_config).not.toBe( + config, + ); + // the caller's object must not be mutated in place + expect(config.watch).toBe(true); + }); + + it('leaves last_get_or_create_request_config undefined if it was never set', () => { + expect( + feed.currentState.last_get_or_create_request_config, + ).toBeUndefined(); + + expect(() => + handleWatchStopped.call(feed, { clearWatchIntent: true }), + ).not.toThrow(); + + expect(feed.currentState.watch).toBe(false); + expect( + feed.currentState.last_get_or_create_request_config, + ).toBeUndefined(); + }); + }); +}); diff --git a/packages/feeds-client/src/feed/event-handlers/watch/handle-watch-stopped.ts b/packages/feeds-client/src/feed/event-handlers/watch/handle-watch-stopped.ts index 84b110ff..31fa035b 100644 --- a/packages/feeds-client/src/feed/event-handlers/watch/handle-watch-stopped.ts +++ b/packages/feeds-client/src/feed/event-handlers/watch/handle-watch-stopped.ts @@ -1,5 +1,31 @@ import type { Feed } from '../../feed'; -export function handleWatchStopped(this: Feed) { - this.state.partialNext({ watch: false }); +export function handleWatchStopped( + this: Feed, + { + clearWatchIntent = false, + }: { + /** + * Also clears the watch intent stored in `last_get_or_create_request_config`, + * so the feed is not re-fetched (and silently re-watched) by `synchronize()` + * on the next reconnect. Pass `true` only when the caller explicitly stopped + * watching — a lost connection must keep the intent so recovery can restore it. + */ + clearWatchIntent?: boolean; + } = {}, +) { + if (!clearWatchIntent) { + this.state.partialNext({ watch: false }); + return; + } + + this.state.next((currentState) => ({ + ...currentState, + watch: false, + last_get_or_create_request_config: + currentState.last_get_or_create_request_config && { + ...currentState.last_get_or_create_request_config, + watch: false, + }, + })); } diff --git a/packages/feeds-client/src/feed/feed.test.ts b/packages/feeds-client/src/feed/feed.test.ts index 796f5258..55e89299 100644 --- a/packages/feeds-client/src/feed/feed.test.ts +++ b/packages/feeds-client/src/feed/feed.test.ts @@ -6,6 +6,7 @@ import type { ActivityResponse, GetOrCreateFeedResponse } from '../gen/models'; import { generateActivityResponse, generateFeedResponse } from '../test-utils'; import { clearQueuedFeeds } from '../utils/throttling'; import { StreamApiError, type StreamResponse } from '../common/types'; +import { handleWatchStopped } from './event-handlers/watch/handle-watch-stopped'; describe('Feed derived state updates', () => { let feed: Feed; @@ -728,6 +729,20 @@ describe('synchronize retry', () => { expect(getOrCreateSpy).not.toHaveBeenCalled(); }); + it('should not synchronize after the watch intent is cleared', async () => { + feed.state.partialNext({ + watch: true, + last_get_or_create_request_config: { watch: true, limit: 20 }, + }); + + // what an explicit feed.stopWatching() does to the feed's state + handleWatchStopped.call(feed, { clearWatchIntent: true }); + + await feed.synchronize(); + + expect(getOrCreateSpy).not.toHaveBeenCalled(); + }); + it('should synchronize and retry on failure', async () => { const mockResponse: StreamResponse = { activities: [], diff --git a/packages/feeds-client/src/feeds-client/feeds-client.test.ts b/packages/feeds-client/src/feeds-client/feeds-client.test.ts index 875f1a04..13b6c3b7 100644 --- a/packages/feeds-client/src/feeds-client/feeds-client.test.ts +++ b/packages/feeds-client/src/feeds-client/feeds-client.test.ts @@ -888,3 +888,142 @@ describe('Feeds client tests', () => { }); }); }); + +describe('reconnect reconciliation', () => { + let client: FeedsClient; + + // FeedsClient extends the generated FeedsApi; stubbing the grandparent + // prototype intercepts the actual network calls. + const feedsApiPrototype = () => + Object.getPrototypeOf(Object.getPrototypeOf(client)); + + const flush = () => new Promise((resolve) => setTimeout(resolve, 0)); + + const reconnect = async () => { + client['eventDispatcher'].dispatch({ + type: 'connection.changed', + online: false, + }); + client['eventDispatcher'].dispatch({ + type: 'connection.changed', + online: true, + }); + await flush(); + }; + + beforeEach(() => { + client = new FeedsClient('mock-api-key'); + + vi.spyOn( + client['connectionIdManager'], + 'getConnectionId', + ).mockResolvedValue('connection-id'); + vi.spyOn(feedsApiPrototype(), 'getOrCreateFeed').mockResolvedValue({ + activities: [], + aggregated_activities: [], + members: [], + next: undefined, + prev: undefined, + duration: '10ms', + feed: generateFeedResponse({ id: 'main', group_id: 'user' }), + }); + vi.spyOn(feedsApiPrototype(), 'stopWatchingFeed').mockResolvedValue({ + duration: '10ms', + }); + vi.spyOn(feedsApiPrototype(), '_queryFeeds').mockResolvedValue({ + feeds: [generateFeedResponse({ id: 'main', group_id: 'user' })], + next: undefined, + prev: undefined, + duration: '10ms', + }); + + // recoverOnReconnect deliberately skips the very first healthy event + client['healthyConnectionChangedEventCount'] = 1; + }); + + afterEach(() => { + vi.restoreAllMocks(); + }); + + it('refetches a feed that is still being watched', async () => { + const feed = client.feed('user', 'main'); + await feed.getOrCreate({ watch: true }); + + const getOrCreateSpy = vi.spyOn(feed, 'getOrCreate'); + + await reconnect(); + + expect(getOrCreateSpy).toHaveBeenCalledWith( + expect.objectContaining({ watch: true }), + ); + }); + + it('does not refetch a feed after stopWatching', async () => { + const feed = client.feed('user', 'main'); + await feed.getOrCreate({ watch: true }); + await feed.stopWatching(); + + expect(feed.currentState.watch).toBe(false); + + const getOrCreateSpy = vi.spyOn(feed, 'getOrCreate'); + + await reconnect(); + + expect(getOrCreateSpy).not.toHaveBeenCalled(); + }); + + it('refetches again once the feed is re-watched after stopWatching', async () => { + const feed = client.feed('user', 'main'); + await feed.getOrCreate({ watch: true }); + await feed.stopWatching(); + await feed.getOrCreate({ watch: true }); + + const getOrCreateSpy = vi.spyOn(feed, 'getOrCreate'); + + await reconnect(); + + expect(getOrCreateSpy).toHaveBeenCalledWith( + expect.objectContaining({ watch: true }), + ); + }); + + it('refetches again when queryFeeds re-watches a stop-watched feed', async () => { + const feed = client.feed('user', 'main'); + await feed.getOrCreate({ watch: true }); + await feed.stopWatching(); + + await client.queryFeeds({ filter: { feed: 'user:main' }, watch: true }); + + // state.watch and the replay gate must not disagree + expect(feed.currentState.watch).toBe(true); + expect(feed.currentState.last_get_or_create_request_config?.watch).toBe( + true, + ); + + const getOrCreateSpy = vi.spyOn(feed, 'getOrCreate'); + + await reconnect(); + + expect(getOrCreateSpy).toHaveBeenCalledWith( + expect.objectContaining({ watch: true }), + ); + }); + + it('only stops refetching the feed that was stop-watched', async () => { + const stopped = client.feed('user', 'stopped'); + const watched = client.feed('user', 'watched'); + await stopped.getOrCreate({ watch: true }); + await watched.getOrCreate({ watch: true }); + await stopped.stopWatching(); + + const stoppedSpy = vi.spyOn(stopped, 'getOrCreate'); + const watchedSpy = vi.spyOn(watched, 'getOrCreate'); + + await reconnect(); + + expect(stoppedSpy).not.toHaveBeenCalled(); + expect(watchedSpy).toHaveBeenCalledWith( + expect.objectContaining({ watch: true }), + ); + }); +}); diff --git a/packages/feeds-client/src/feeds-client/feeds-client.ts b/packages/feeds-client/src/feeds-client/feeds-client.ts index ecae7e53..c1e57472 100644 --- a/packages/feeds-client/src/feeds-client/feeds-client.ts +++ b/packages/feeds-client/src/feeds-client/feeds-client.ts @@ -1229,7 +1229,9 @@ export class FeedsClient extends FeedsApi { const feeds = this.findAllActiveFeedsByFid( `${request.feed_group_id}:${request.feed_id}`, ); - feeds.forEach((f) => handleWatchStopped.bind(f)()); + feeds.forEach((f) => + handleWatchStopped.call(f, { clearWatchIntent: true }), + ); return response; } @@ -1247,7 +1249,9 @@ export class FeedsClient extends FeedsApi { const feeds = this.findAllActiveFeedsByFid( `${request.feed_group_id}:${request.feed_id}`, ); - feeds.forEach((f) => handleWatchStarted.bind(f)()); + feeds.forEach((f) => + handleWatchStarted.call(f, { setWatchIntent: true }), + ); } return response; @@ -1326,7 +1330,7 @@ export class FeedsClient extends FeedsApi { wasFullUpdate = true; } } - if (watch) handleWatchStarted.call(feed); + if (watch) handleWatchStarted.call(feed, { setWatchIntent: true }); } // If we didn't do a full update, check if own_* fields have changed