diff --git a/packages/assets-controller/src/AssetsController.test.ts b/packages/assets-controller/src/AssetsController.test.ts index f8467cf50c..cf9460b428 100644 --- a/packages/assets-controller/src/AssetsController.test.ts +++ b/packages/assets-controller/src/AssetsController.test.ts @@ -85,6 +85,12 @@ type AllEvents = MessengerEvents; type RootMessenger = Messenger; +/** Mirrors the private `RESUBSCRIBE_DEBOUNCE_MS` in AssetsController.ts. */ +const RESUBSCRIBE_DEBOUNCE_MS = 250; + +/** Mirrors the private `RESUBSCRIBE_JITTER_MS` in AssetsController.ts. */ +const RESUBSCRIBE_JITTER_MS = 5000; + const MOCK_ACCOUNT_ID = 'mock-account-id-1'; const MOCK_ASSET_ID = 'eip155:1/erc20:0xA0b86991c6218b36c1d19D4a2e9Eb0cE3606eB48' as Caip19AssetId; @@ -1392,14 +1398,81 @@ describe('AssetsController', () => { }); describe('handleActiveChainsUpdate', () => { - it('re-subscribes assets when chains are added', async () => { + it('re-subscribes assets when chains are added, debounced and jittered', async () => { await withController(async ({ controller }) => { - const subscribeSpy = jest.spyOn(controller, 'subscribeAssetsPrice'); + jest.useFakeTimers(); + const randomSpy = jest.spyOn(Math, 'random').mockReturnValue(0.5); + try { + const subscribeSpy = jest.spyOn(controller, 'subscribeAssetsPrice'); - const onActiveChainsUpdated = controller.getOnActiveChainsUpdated(); - onActiveChainsUpdated('TestDataSource', ['eip155:1'], []); + const onActiveChainsUpdated = controller.getOnActiveChainsUpdated(); + onActiveChainsUpdated('TestDataSource', ['eip155:1'], []); + + // Re-subscribe is debounced, so it does not run synchronously. + expect(subscribeSpy).not.toHaveBeenCalled(); - expect(subscribeSpy).toHaveBeenCalledTimes(1); + // Additions are jittered on top of the base debounce, so nothing runs + // at the base debounce window. + jest.advanceTimersByTime(RESUBSCRIBE_DEBOUNCE_MS); + expect(subscribeSpy).not.toHaveBeenCalled(); + + // Once the jitter window elapses it runs exactly once. + jest.advanceTimersByTime(RESUBSCRIBE_JITTER_MS); + expect(subscribeSpy).toHaveBeenCalledTimes(1); + } finally { + randomSpy.mockRestore(); + jest.useRealTimers(); + } + }); + }); + + it('re-subscribes promptly (no jitter) when a chain removal pre-empts a pending jittered addition', async () => { + await withController(async ({ controller }) => { + jest.useFakeTimers(); + const randomSpy = jest.spyOn(Math, 'random').mockReturnValue(0.9); + try { + const subscribeSpy = jest.spyOn(controller, 'subscribeAssetsPrice'); + + const onActiveChainsUpdated = controller.getOnActiveChainsUpdated(); + // Addition schedules a jittered re-subscribe... + onActiveChainsUpdated('TestDataSource', ['eip155:1'], []); + // ...then a removal arrives; it must pre-empt to the prompt window. + onActiveChainsUpdated('TestDataSource', [], ['eip155:137']); + + jest.advanceTimersByTime(RESUBSCRIBE_DEBOUNCE_MS); + expect(subscribeSpy).toHaveBeenCalledTimes(1); + } finally { + randomSpy.mockRestore(); + jest.useRealTimers(); + } + }); + }); + + it('coalesces a burst of active-chain updates into a single re-subscribe', async () => { + await withController(async ({ controller }) => { + jest.useFakeTimers(); + try { + const subscribeSpy = jest.spyOn(controller, 'subscribeAssetsPrice'); + + const onActiveChainsUpdated = controller.getOnActiveChainsUpdated(); + // Simulate a burst of chain up/down notifications within the window. + onActiveChainsUpdated('TestDataSource', ['eip155:1'], []); + onActiveChainsUpdated( + 'TestDataSource', + ['eip155:1', 'eip155:137'], + ['eip155:1'], + ); + onActiveChainsUpdated( + 'TestDataSource', + ['eip155:137'], + ['eip155:1', 'eip155:137'], + ); + + jest.advanceTimersByTime(RESUBSCRIBE_DEBOUNCE_MS); + expect(subscribeSpy).toHaveBeenCalledTimes(1); + } finally { + jest.useRealTimers(); + } }); }); @@ -1428,12 +1501,18 @@ describe('AssetsController', () => { it('re-subscribes assets when chains are removed', async () => { await withController(async ({ controller }) => { - const subscribeSpy = jest.spyOn(controller, 'subscribeAssetsPrice'); + jest.useFakeTimers(); + try { + const subscribeSpy = jest.spyOn(controller, 'subscribeAssetsPrice'); - const onActiveChainsUpdated = controller.getOnActiveChainsUpdated(); - onActiveChainsUpdated('TestDataSource', [], ['eip155:1']); + const onActiveChainsUpdated = controller.getOnActiveChainsUpdated(); + onActiveChainsUpdated('TestDataSource', [], ['eip155:1']); - expect(subscribeSpy).toHaveBeenCalledTimes(1); + jest.advanceTimersByTime(RESUBSCRIBE_DEBOUNCE_MS); + expect(subscribeSpy).toHaveBeenCalledTimes(1); + } finally { + jest.useRealTimers(); + } }); }); @@ -2019,8 +2098,19 @@ describe('AssetsController', () => { }; await withController( - { state: initialState }, + { state: initialState, clientControllerState: { isUiOpen: true } }, async ({ controller, messenger }) => { + // UI must be open and keyring unlocked so AccountActivityDataSource is + // subscribed and can route balanceUpdated events to the account. + ( + messenger as unknown as { + publish: (topic: string, payload?: unknown) => void; + } + ).publish('ClientController:stateChange', { isUiOpen: true }); + messenger.publish('KeyringController:unlock'); + + await flushPromises(); + messenger.publish('AccountActivityService:balanceUpdated', { address: '0x1234567890123456789012345678901234567890', chain: 'eip155:42161', diff --git a/packages/assets-controller/src/AssetsController.ts b/packages/assets-controller/src/AssetsController.ts index b6526f4da2..99c9aca096 100644 --- a/packages/assets-controller/src/AssetsController.ts +++ b/packages/assets-controller/src/AssetsController.ts @@ -16,9 +16,9 @@ import type { TraceCallback } from '@metamask/controller-utils'; import type { ApiPlatformClient, AccountActivityServiceBalanceUpdatedEvent, + AccountActivityServiceStatusChangedEvent, BackendWebSocketServiceActions, BackendWebSocketServiceEvents, - BalanceUpdate, SupportedCurrency, } from '@metamask/core-backend'; import type { @@ -77,6 +77,7 @@ import type { } from './data-sources/AbstractDataSource'; import type { AccountsApiDataSourceConfig } from './data-sources/AccountsApiDataSource'; import { AccountsApiDataSource } from './data-sources/AccountsApiDataSource'; +import { AccountActivityDataSource } from './data-sources/AccountActivityDataSource'; import { BackendWebsocketDataSource } from './data-sources/BackendWebsocketDataSource'; import { shouldSkipNativeForCaipChainId } from './data-sources/evm-rpc-services/utils/assets'; import type { PriceDataSourceConfig } from './data-sources/PriceDataSource'; @@ -146,7 +147,6 @@ import type { } from './utils'; import { ZERO_ADDRESS } from './utils/constants'; import { pickRpcCustomAssetsSupplement } from './utils/customAssetsRpcSupplement'; -import { processAccountActivityBalanceUpdates } from './utils/processAccountActivityBalanceUpdates'; const NATIVE_ASSETS_QUERY_KEY = ['nativeAssets']; @@ -195,6 +195,26 @@ const MESSENGER_EXPOSED_METHODS = [ /** Default polling interval hint for data sources (30 seconds) */ const DEFAULT_POLLING_INTERVAL_MS = 30_000; +/** + * Trailing debounce window (ms) for coalescing event-driven re-subscribes. + * A burst of `onActiveChainsUpdated` callbacks (e.g. many `statusChanged` + * up/down notifications, or several data sources reporting supported chains at + * init) collapses into a single re-subscribe pass so polling data sources are + * not torn down and re-created (with a fresh immediate fetch) once per event. + */ +const RESUBSCRIBE_DEBOUNCE_MS = 250; + +/** + * Maximum random jitter (ms) added to the re-subscribe delay when a data source + * *claims* new chains (e.g. WebSocket `statusChanged: 'up'`). Staggers the + * resulting WS-subscribe fan-out across clients that all receive the same + * backend broadcast, avoiding a thundering herd. Only applied to chain + * *additions*: additions are non-urgent (polling still covers the chain until + * re-subscribe), whereas removals are re-subscribed promptly so polling fallback + * resumes without a data gap. + */ +const RESUBSCRIBE_JITTER_MS = 5000; + // ============================================================================ // TRACE NAMES — used in Sentry spans (search these strings in Discover) // ============================================================================ @@ -348,8 +368,9 @@ type AllowedEvents = | SnapControllerSnapInstalledEvent // BackendWebsocketDataSource | BackendWebSocketServiceEvents - // AccountActivityService (real-time balance updates for unified assets) - | AccountActivityServiceBalanceUpdatedEvent; + // AccountActivityService (real-time balance updates + chain status for unified assets) + | AccountActivityServiceBalanceUpdatedEvent + | AccountActivityServiceStatusChangedEvent; export type AssetsControllerMessenger = Messenger< typeof CONTROLLER_NAME, @@ -741,6 +762,20 @@ export class AssetsController extends BaseController< */ readonly #activeSubscriptions: Map = new Map(); + /** + * Pending trailing-debounce timer for coalesced re-subscribes. Bursts of + * event-driven `#scheduleSubscribeAssets()` calls collapse into a single + * `#subscribeAssets()` run. See {@link RESUBSCRIBE_DEBOUNCE_MS}. + */ + #resubscribeTimer: ReturnType | null = null; + + /** + * Scheduled fire time (epoch ms) of {@link #resubscribeTimer}. Lets an urgent + * (non-jittered) re-subscribe pre-empt a pending jittered one so a chain + * removal is not delayed behind a chain addition's jitter window. + */ + #resubscribeRunAt = 0; + /** Currently enabled chains from NetworkEnablementController */ #enabledChains: Set = new Set(); @@ -776,6 +811,8 @@ export class AssetsController extends BaseController< readonly #backendWebsocketDataSource: BackendWebsocketDataSource; + readonly #accountActivityDataSource: AccountActivityDataSource; + readonly #accountsApiDataSource: AccountsApiDataSource; readonly #snapDataSource: SnapDataSource; @@ -896,6 +933,12 @@ export class AssetsController extends BaseController< getAssetType: (assetId: Caip19AssetId): 'native' | 'erc20' | 'spl' => this.#getAssetType(assetId), }); + this.#accountActivityDataSource = new AccountActivityDataSource({ + messenger: this.messenger, + onActiveChainsUpdated: this.#onActiveChainsUpdated, + getAssetType: (assetId: Caip19AssetId): 'native' | 'erc20' | 'spl' => + this.#getAssetType(assetId), + }); this.#accountsApiDataSource = new AccountsApiDataSource({ queryApiClient, onActiveChainsUpdated: this.#onActiveChainsUpdated, @@ -1172,15 +1215,8 @@ export class AssetsController extends BaseController< }, ); - // Real-time post-tx balances from AccountActivityService (same WS payload as - // TokenBalancesController; BackendWebsocketDataSource may not receive the - // callback when AccountActivityService owns the server subscription). - this.messenger.subscribe( - 'AccountActivityService:balanceUpdated', - (event) => { - this.#onAccountActivityBalanceUpdated(event); - }, - ); + // Real-time post-tx balances and chain status from AccountActivityService are + // consumed by AccountActivityDataSource (subscribed in #subscribeAssets). } #onUnapprovedTransactionAdded(transactionMeta: TransactionMeta): void { @@ -1239,49 +1275,6 @@ export class AssetsController extends BaseController< }); } - #onAccountActivityBalanceUpdated({ - address, - chain, - updates, - }: { - address: string; - chain: string; - updates: BalanceUpdate[]; - }): void { - const account = this.#getSelectedAccounts().find((a) => - a.address.startsWith('0x') - ? a.address.toLowerCase() === address.toLowerCase() - : a.address === address, - ); - - if (!account) { - return; - } - - const chainId = chain as ChainId; - const response = processAccountActivityBalanceUpdates( - updates, - account.id, - (assetId) => this.#getAssetType(assetId), - ); - - if (!response.assetsBalance) { - return; - } - - const request: DataRequest = { - accountsWithSupportedChains: [{ account, supportedChains: [chainId] }], - chainIds: [chainId], - dataTypes: ['balance', 'metadata'], - }; - - this.handleAssetsUpdate(response, 'AccountActivityService', request).catch( - (error) => { - log('Failed to apply AccountActivityService balance update', { error }); - }, - ); - } - /** * Start or stop asset tracking based on client (UI) open state and keyring * unlock state. Only runs when both UI is open and keyring is unlocked. @@ -1442,12 +1435,23 @@ export class AssetsController extends BaseController< const addedChains = activeChains.filter((ch) => !previousSet.has(ch)); const removedChains = previous.filter((ch) => !activeChains.includes(ch)); - if (addedChains.length > 0 || removedChains.length > 0) { - // Refresh subscriptions to use updated data source availability. - // No one-time fetch needed here — #handleEnabledNetworksChanged - // handles fetches when the user enables a network, and - // #subscribeAssets re-subscribes with the correct chain assignment. - this.#subscribeAssets(); + // Refresh subscriptions to use updated data source availability. + // No one-time fetch needed here — #handleEnabledNetworksChanged + // handles fetches when the user enables a network, and + // #subscribeAssets re-subscribes with the correct chain assignment. + // Coalesce bursts of chain updates into a single re-subscribe so polling + // data sources are not torn down/re-created (with a redundant immediate + // fetch) once per event. + if (removedChains.length > 0) { + // Urgent: a data source dropped chains (e.g. WebSocket went down). Until + // re-subscribe hands those chains to a polling source there is a data gap, + // so run promptly without jitter. + this.#scheduleSubscribeAssets(); + } else if (addedChains.length > 0) { + // Non-urgent: a data source claimed new chains (polling still covers them + // until re-subscribe). Add jitter to stagger the resulting WS-subscribe + // fan-out across clients receiving the same backend broadcast. + this.#scheduleSubscribeAssets({ jitter: true }); } } @@ -2863,6 +2867,13 @@ export class AssetsController extends BaseController< this.#stateSizeReported = false; this.#lastKnownAccountIds = new Set(); + // Cancel any pending coalesced re-subscribe so it cannot fire after stop. + if (this.#resubscribeTimer) { + clearTimeout(this.#resubscribeTimer); + this.#resubscribeTimer = null; + } + this.#resubscribeRunAt = 0; + // Stop price subscription first (uses direct messenger call) this.unsubscribeAssetsPrice(); @@ -2872,6 +2883,7 @@ export class AssetsController extends BaseController< // every source that may have been subscribed. const allSources = [ ...this.#allBalanceDataSources, + this.#accountActivityDataSource, this.#stakedBalanceDataSource, ]; const subscriptionKeys = [...this.#activeSubscriptions.keys()]; @@ -2908,6 +2920,53 @@ export class AssetsController extends BaseController< this.#subscribeAssets(); } + /** + * Schedule a coalesced re-subscribe. Multiple calls within + * {@link RESUBSCRIBE_DEBOUNCE_MS} collapse into a single `#subscribeAssets()` + * run. Used for event-driven bursts (e.g. `onActiveChainsUpdated` from many + * `statusChanged` notifications) so polling data sources are not repeatedly + * torn down and re-created (each re-create firing an immediate, redundant + * fetch). Explicit flows (startup, account changes, enabled-network changes) + * keep calling `#subscribeAssets()` directly so their re-subscribe is not + * delayed. + * + * When `jitter` is set, a random delay up to {@link RESUBSCRIBE_JITTER_MS} is + * added to stagger the WS-subscribe fan-out across clients (used for chain + * additions). A later non-jittered call pre-empts a pending jittered one so an + * urgent chain removal is not held back by an addition's jitter window. + * + * @param options - Scheduling options. + * @param options.jitter - Add random jitter to the delay (for chain additions). + */ + #scheduleSubscribeAssets({ jitter = false }: { jitter?: boolean } = {}): void { + const delay = jitter + ? RESUBSCRIBE_DEBOUNCE_MS + Math.floor(Math.random() * RESUBSCRIBE_JITTER_MS) + : RESUBSCRIBE_DEBOUNCE_MS; + const runAt = Date.now() + delay; + + if (this.#resubscribeTimer) { + // Keep the earliest scheduled run so an urgent (shorter-delay) request can + // pre-empt a pending jittered one, but a jittered request never pushes a + // sooner pending run later. + if (runAt >= this.#resubscribeRunAt) { + return; + } + clearTimeout(this.#resubscribeTimer); + this.#resubscribeTimer = null; + } + + this.#resubscribeRunAt = runAt; + this.#resubscribeTimer = setTimeout(() => { + this.#resubscribeTimer = null; + this.#resubscribeRunAt = 0; + try { + this.#subscribeAssets(); + } catch (error) { + log('Failed to run coalesced re-subscribe', error); + } + }, delay); + } + /** * Subscribe to asset updates for all selected accounts. */ @@ -3004,6 +3063,17 @@ export class AssetsController extends BaseController< chainToAccounts, rpcAssignedChains, ); + + // AccountActivityDataSource consumes real-time AccountActivityService events. + // It does not participate in chain-claiming; it is subscribed with all + // selected accounts and enabled chains so it can route `balanceUpdated` + // events to the matching account. Chain up/down is managed internally from + // `statusChanged` (see AccountActivityDataSource). + this.#subscribeDataSource( + this.#accountActivityDataSource, + accounts, + chainIds, + ); } /** @@ -3602,7 +3672,7 @@ export class AssetsController extends BaseController< const shouldGraduateCustomAssets = sourceId === 'AccountsApiDataSource' || sourceId === 'BackendWebsocketDataSource' || - sourceId === 'AccountActivityService'; + sourceId === 'AccountActivityDataSource'; const enrichmentSources: AssetsDataSource[] = [ ...(shouldGraduateCustomAssets @@ -3654,6 +3724,7 @@ export class AssetsController extends BaseController< // Destroy instantiated data sources this.#backendWebsocketDataSource?.destroy?.(); + this.#accountActivityDataSource?.destroy?.(); this.#accountsApiDataSource?.destroy?.(); this.#snapDataSource?.destroy?.(); this.#rpcDataSource?.destroy?.(); diff --git a/packages/assets-controller/src/data-sources/AccountActivityDataSource.ts b/packages/assets-controller/src/data-sources/AccountActivityDataSource.ts new file mode 100644 index 0000000000..6bb227b48f --- /dev/null +++ b/packages/assets-controller/src/data-sources/AccountActivityDataSource.ts @@ -0,0 +1,342 @@ +import type { + AccountActivityServiceBalanceUpdatedEvent, + AccountActivityServiceStatusChangedEvent, + BalanceUpdate, +} from '@metamask/core-backend'; +import type { InternalAccount } from '@metamask/keyring-internal-api'; +import { isCaipChainId } from '@metamask/utils'; + +import type { AssetsControllerMessenger } from '../AssetsController'; +import { projectLogger, createModuleLogger } from '../logger'; +import type { ChainId, Caip19AssetId, DataRequest } from '../types'; +import { processAccountActivityBalanceUpdates } from '../utils/processAccountActivityBalanceUpdates'; +import { AbstractDataSource } from './AbstractDataSource'; +import type { + DataSourceState, + SubscriptionRequest, +} from './AbstractDataSource'; + +// ============================================================================ +// CONSTANTS +// ============================================================================ + +const CONTROLLER_NAME = 'AccountActivityDataSource'; + +const log = createModuleLogger(projectLogger, CONTROLLER_NAME); + +// ============================================================================ +// MESSENGER TYPES +// ============================================================================ + +/** Allowed events that AccountActivityDataSource subscribes to. */ +export type AccountActivityDataSourceAllowedEvents = + | AccountActivityServiceBalanceUpdatedEvent + | AccountActivityServiceStatusChangedEvent; + +// ============================================================================ +// STATE +// ============================================================================ + +export type AccountActivityDataSourceState = DataSourceState; + +const defaultState: AccountActivityDataSourceState = { + activeChains: [], +}; + +// ============================================================================ +// OPTIONS +// ============================================================================ + +export type AccountActivityDataSourceOptions = { + /** The AssetsController messenger (shared by all data sources). */ + messenger: AssetsControllerMessenger; + /** Called when active chains are updated. Pass dataSourceName so the controller knows the source. */ + onActiveChainsUpdated: ( + dataSourceName: string, + chains: ChainId[], + previousChains: ChainId[], + ) => void; + /** Returns the asset type ('native' | 'erc20' | 'spl') for a given CAIP-19 asset ID. */ + getAssetType: (assetId: Caip19AssetId) => 'native' | 'erc20' | 'spl'; + state?: Partial; +}; + +// ============================================================================ +// ACCOUNT ACTIVITY DATA SOURCE +// ============================================================================ + +/** + * Data source that consumes real-time updates from `AccountActivityService`. + * + * Unlike {@link BackendWebsocketDataSource}, which owns its own WebSocket + * channel subscriptions, this data source is a thin consumer of the two + * high-level events that `AccountActivityService` publishes: + * + * - `AccountActivityService:balanceUpdated` — post-transaction balances for the + * subscribed account(s). The address is resolved against the accounts in the + * active subscription(s), transformed into a {@link DataResponse} (via + * {@link processAccountActivityBalanceUpdates}), and pushed to the controller + * through the subscription's `onAssetsUpdate` callback. + * - `AccountActivityService:statusChanged` — per-chain "up"/"down" notifications. + * A chain reported "up" is claimed as an active chain (WebSocket is providing + * real-time data); a chain reported "down" is released so polling data sources + * can take over. Each change is applied directly via `updateActiveChains`, + * which notifies the controller through `onActiveChainsUpdated`. + * + * This data source does NOT debounce, jitter, or gate chain updates itself. + * Coalescing (collapsing bursts) and jitter (staggering the WS-subscribe herd + * across clients) live in `AssetsController`, where the expensive re-subscribe + * happens and where updates from all data sources converge. `activeChains` also + * are never seeded from the accounts API the way `BackendWebsocketDataSource` + * does; they only ever reflect live `statusChanged` events (the service flushes + * all tracked chains as "down" when the WebSocket disconnects, releasing them). + * + * It subscribes to these events in its constructor (the same way + * `TokenBalancesController` does) and is wired into the controller's balance + * subscription flow so incoming balance updates can be routed to the right + * account. + */ +export class AccountActivityDataSource extends AbstractDataSource< + typeof CONTROLLER_NAME, + AccountActivityDataSourceState +> { + readonly #messenger: AssetsControllerMessenger; + + readonly #onActiveChainsUpdated: ( + dataSourceName: string, + chains: ChainId[], + previousChains: ChainId[], + ) => void; + + readonly #getAssetType: ( + assetId: Caip19AssetId, + ) => 'native' | 'erc20' | 'spl'; + + /** Stored subscription requests (used to route incoming balance updates). */ + readonly #subscriptionRequests: Map = new Map(); + + /** Unsubscribe handles for messenger event subscriptions. */ + readonly #eventUnsubscribes: (() => void)[] = []; + + constructor(options: AccountActivityDataSourceOptions) { + super(CONTROLLER_NAME, { + ...defaultState, + ...options.state, + }); + + this.#messenger = options.messenger; + this.#onActiveChainsUpdated = options.onActiveChainsUpdated; + this.#getAssetType = options.getAssetType; + + this.#subscribeToEvents(); + } + + // ============================================================================ + // EVENT SUBSCRIPTIONS + // ============================================================================ + + #subscribeToEvents(): void { + const unsubscribeBalance = this.#messenger.subscribe( + 'AccountActivityService:balanceUpdated', + (event) => this.#onBalanceUpdated(event), + ); + const unsubscribeStatus = this.#messenger.subscribe( + 'AccountActivityService:statusChanged', + this.#onAccountActivityStatusChanged, + ); + + if (typeof unsubscribeBalance === 'function') { + this.#eventUnsubscribes.push(unsubscribeBalance); + } + if (typeof unsubscribeStatus === 'function') { + this.#eventUnsubscribes.push(unsubscribeStatus); + } + } + + // ============================================================================ + // SUBSCRIBE / UNSUBSCRIBE + // ============================================================================ + + async subscribe(subscriptionRequest: SubscriptionRequest): Promise { + const { subscriptionId, request } = subscriptionRequest; + + // Store the request so incoming `balanceUpdated` events can be routed to the + // matching account and reported through its `onAssetsUpdate` callback. + this.#subscriptionRequests.set(subscriptionId, subscriptionRequest); + this.activeSubscriptions.set(subscriptionId, { + cleanup: () => { + this.#subscriptionRequests.delete(subscriptionId); + }, + chains: request.chainIds, + addresses: request.accountsWithSupportedChains.map( + (entry) => entry.account.address, + ), + onAssetsUpdate: subscriptionRequest.onAssetsUpdate, + }); + } + + // ============================================================================ + // BALANCE UPDATES + // ============================================================================ + + #onBalanceUpdated({ + address, + chain, + updates, + }: { + address: string; + chain: string; + updates: BalanceUpdate[]; + }): void { + try { + if (!address || !chain || !updates || updates.length === 0) { + return; + } + + const match = this.#findAccountForAddress(address); + if (!match) { + return; + } + + const { account, onAssetsUpdate } = match; + const chainId = chain as ChainId; + + const response = processAccountActivityBalanceUpdates( + updates, + account.id, + (assetId) => this.#getAssetType(assetId), + ); + + if (!response.assetsBalance) { + return; + } + + const request: DataRequest = { + accountsWithSupportedChains: [{ account, supportedChains: [chainId] }], + chainIds: [chainId], + dataTypes: ['balance', 'metadata'], + }; + + Promise.resolve(onAssetsUpdate(response, request)).catch((error) => { + log('Failed to report balance update', { error }); + }); + } catch (error) { + log('Error handling balance update', error); + } + } + + /** + * Find the internal account matching the given activity address across all + * active subscription requests, along with the callback used to report its + * updates. EVM addresses are matched case-insensitively; other namespaces are + * matched exactly. + * + * @param address - The account address from the activity message. + * @returns The matching account and its `onAssetsUpdate` callback, or null. + */ + #findAccountForAddress(address: string): { + account: InternalAccount; + onAssetsUpdate: SubscriptionRequest['onAssetsUpdate']; + } | null { + const isEvm = address.startsWith('0x'); + + for (const subscriptionRequest of this.#subscriptionRequests.values()) { + const account = subscriptionRequest.request.accountsWithSupportedChains + .map((entry) => entry.account) + .find((candidate) => + isEvm + ? candidate.address.toLowerCase() === address.toLowerCase() + : candidate.address === address, + ); + + if (account) { + return { + account, + onAssetsUpdate: subscriptionRequest.onAssetsUpdate, + }; + } + } + + return null; + } + + // ============================================================================ + // STATUS CHANGES + // ============================================================================ + + /** + * Handle a `statusChanged` notification. Chains reported "up" are claimed as + * active (WebSocket provides real-time data); chains reported "down" are + * released so polling data sources take over. The change is applied directly; + * coalescing and jitter are handled by `AssetsController`. + */ + readonly #onAccountActivityStatusChanged = ({ + chainIds, + status, + }: { + chainIds: string[]; + status: 'up' | 'down'; + timestamp?: number; + }): void => { + try { + // Act on every namespace (eip155, solana, etc.); AssetsController is + // multichain. Only skip identifiers that are not valid CAIP-2 chain IDs. + const validChains = chainIds.filter((chainId) => + isCaipChainId(chainId), + ) as ChainId[]; + + if (validChains.length === 0) { + return; + } + + const next = new Set(this.state.activeChains); + if (status === 'up') { + for (const chainId of validChains) { + next.add(chainId); + } + } else { + for (const chainId of validChains) { + next.delete(chainId); + } + } + + const previous = [...this.state.activeChains]; + this.updateActiveChains(Array.from(next), (updatedChains) => + this.#onActiveChainsUpdated(this.getName(), updatedChains, previous), + ); + } catch (error) { + log('Error handling status change', error); + } + }; + + // ============================================================================ + // CLEANUP + // ============================================================================ + + destroy(): void { + for (const unsubscribe of this.#eventUnsubscribes) { + unsubscribe(); + } + this.#eventUnsubscribes.length = 0; + + this.#subscriptionRequests.clear(); + + super.destroy(); + } +} + +// ============================================================================ +// FACTORY FUNCTION +// ============================================================================ + +/** + * Creates an AccountActivityDataSource instance. + * + * @param options - Configuration options for the data source. + * @returns A new AccountActivityDataSource instance. + */ +export function createAccountActivityDataSource( + options: AccountActivityDataSourceOptions, +): AccountActivityDataSource { + return new AccountActivityDataSource(options); +} diff --git a/packages/assets-controller/src/data-sources/AccountsApiDataSource.test.ts b/packages/assets-controller/src/data-sources/AccountsApiDataSource.test.ts index 645b0a44de..7161fb79b9 100644 --- a/packages/assets-controller/src/data-sources/AccountsApiDataSource.test.ts +++ b/packages/assets-controller/src/data-sources/AccountsApiDataSource.test.ts @@ -512,6 +512,135 @@ describe('AccountsApiDataSource', () => { controller.destroy(); }); + it('skips the immediate fetch when re-subscribing with an unchanged scope within one poll interval', async () => { + const { controller, apiClient, assetsUpdateHandler } = + await setupController(); + + await controller.subscribe({ + subscriptionId: 'sub-1', + request: createDataRequest(), + isUpdate: false, + onAssetsUpdate: assetsUpdateHandler, + }); + + expect( + apiClient.accounts.fetchV5MultiAccountBalances, + ).toHaveBeenCalledTimes(1); + expect(assetsUpdateHandler).toHaveBeenCalledTimes(1); + + // Re-subscribe with the identical scope (e.g. a chain flapped back from the + // WebSocket source). The immediate fetch is skipped as redundant. + await controller.subscribe({ + subscriptionId: 'sub-1', + request: createDataRequest(), + isUpdate: false, + onAssetsUpdate: assetsUpdateHandler, + }); + + expect( + apiClient.accounts.fetchV5MultiAccountBalances, + ).toHaveBeenCalledTimes(1); + expect(assetsUpdateHandler).toHaveBeenCalledTimes(1); + + controller.destroy(); + }); + + it('performs the immediate fetch when re-subscribing with forceUpdate even if the scope is unchanged', async () => { + const { controller, apiClient, assetsUpdateHandler } = + await setupController(); + + await controller.subscribe({ + subscriptionId: 'sub-1', + request: createDataRequest(), + isUpdate: false, + onAssetsUpdate: assetsUpdateHandler, + }); + + expect( + apiClient.accounts.fetchV5MultiAccountBalances, + ).toHaveBeenCalledTimes(1); + + await controller.subscribe({ + subscriptionId: 'sub-1', + request: createDataRequest({ forceUpdate: true }), + isUpdate: false, + onAssetsUpdate: assetsUpdateHandler, + }); + + expect( + apiClient.accounts.fetchV5MultiAccountBalances, + ).toHaveBeenCalledTimes(2); + + controller.destroy(); + }); + + it('performs the immediate fetch when re-subscribing with a different scope', async () => { + const { controller, apiClient, assetsUpdateHandler } = + await setupController(); + + await controller.subscribe({ + subscriptionId: 'sub-1', + request: createDataRequest({ chainIds: [CHAIN_MAINNET] }), + isUpdate: false, + onAssetsUpdate: assetsUpdateHandler, + }); + + expect( + apiClient.accounts.fetchV5MultiAccountBalances, + ).toHaveBeenCalledTimes(1); + + await controller.subscribe({ + subscriptionId: 'sub-1', + request: createDataRequest({ chainIds: [CHAIN_MAINNET, CHAIN_POLYGON] }), + isUpdate: false, + onAssetsUpdate: assetsUpdateHandler, + }); + + expect( + apiClient.accounts.fetchV5MultiAccountBalances, + ).toHaveBeenCalledTimes(2); + + controller.destroy(); + }); + + it('performs the immediate fetch when re-subscribing after the previous fetch has gone stale', async () => { + const { controller, apiClient, assetsUpdateHandler } = + await setupController(); + + const nowSpy = jest.spyOn(Date, 'now'); + const startTime = 1_000_000; + nowSpy.mockReturnValue(startTime); + + await controller.subscribe({ + subscriptionId: 'sub-1', + request: createDataRequest(), + isUpdate: false, + onAssetsUpdate: assetsUpdateHandler, + }); + + expect( + apiClient.accounts.fetchV5MultiAccountBalances, + ).toHaveBeenCalledTimes(1); + + // Advance the clock past the default poll interval (30s) so the previous + // fetch is no longer fresh. + nowSpy.mockReturnValue(startTime + 30_001); + + await controller.subscribe({ + subscriptionId: 'sub-1', + request: createDataRequest(), + isUpdate: false, + onAssetsUpdate: assetsUpdateHandler, + }); + + expect( + apiClient.accounts.fetchV5MultiAccountBalances, + ).toHaveBeenCalledTimes(2); + + nowSpy.mockRestore(); + controller.destroy(); + }); + it('subscribe does nothing when no chains', async () => { const { controller, assetsUpdateHandler } = await setupController(); diff --git a/packages/assets-controller/src/data-sources/AccountsApiDataSource.ts b/packages/assets-controller/src/data-sources/AccountsApiDataSource.ts index 74f5d94be1..a6cc65c79a 100644 --- a/packages/assets-controller/src/data-sources/AccountsApiDataSource.ts +++ b/packages/assets-controller/src/data-sources/AccountsApiDataSource.ts @@ -16,7 +16,11 @@ import type { Middleware, AssetsControllerStateInternal, } from '../types'; -import { fetchWithTimeout, normalizeAssetId } from '../utils'; +import { + computeSubscriptionScopeKey, + fetchWithTimeout, + normalizeAssetId, +} from '../utils'; import type { DataSourceState, SubscriptionRequest, @@ -201,6 +205,15 @@ export class AccountsApiDataSource extends AbstractDataSource< /** State accessor from subscriptions (for filtering when tokenDetectionEnabled is false) */ #getAssetsState?: () => AssetsControllerStateInternal; + /** + * Scope key + timestamp of the last fetch, preserved across teardown. Used to + * skip the redundant immediate fetch when the subscription is re-created for + * the same scope while its last fetch is still fresh (e.g. a chain flapping + * between the WebSocket and this source). This source only ever has a single + * subscription, so a single record suffices. + */ + #lastFetch: { scopeKey: string; at: number } | null = null; + constructor(options: AccountsApiDataSourceOptions) { super(CONTROLLER_NAME, { ...defaultState, @@ -564,6 +577,11 @@ export class AccountsApiDataSource extends AbstractDataSource< await this.unsubscribe(subscriptionId); const pollInterval = request.updateInterval ?? this.#pollInterval; + const scopeKey = computeSubscriptionScopeKey( + request, + chainsToSubscribe, + pollInterval, + ); // Create poll function for this subscription const pollFn = async (): Promise => { @@ -573,6 +591,8 @@ export class AccountsApiDataSource extends AbstractDataSource< return; } + this.#lastFetch = { scopeKey, at: Date.now() }; + // Use stored request (which gets updated on account changes) const fetchResponse = await this.fetch({ ...subscription.request, @@ -601,6 +621,24 @@ export class AccountsApiDataSource extends AbstractDataSource< onAssetsUpdate: subscriptionRequest.onAssetsUpdate, }); + // Skip the immediate fetch when this subscription was just re-created for + // the same scope and its last fetch is still fresh (within one poll + // interval). This avoids redundant identical requests when a chain flaps + // between the WebSocket source and this polling source. `forceUpdate` + // always fetches (it is excluded from the scope key). The scheduled poll + // above still refreshes on the next interval tick. + const isRedundantImmediateFetch = + !request.forceUpdate && + this.#lastFetch?.scopeKey === scopeKey && + Date.now() - this.#lastFetch.at < pollInterval; + + if (isRedundantImmediateFetch) { + log('Skipping redundant immediate fetch on re-subscribe', { + subscriptionId, + }); + return; + } + // Initial fetch await pollFn(); } @@ -615,6 +653,8 @@ export class AccountsApiDataSource extends AbstractDataSource< clearInterval(this.#chainsRefreshTimer); } + this.#lastFetch = null; + // Clean up subscriptions super.destroy(); } diff --git a/packages/assets-controller/src/data-sources/RpcDataSource.test.ts b/packages/assets-controller/src/data-sources/RpcDataSource.test.ts index c2ff9044d1..2ae3e8f6b5 100644 --- a/packages/assets-controller/src/data-sources/RpcDataSource.test.ts +++ b/packages/assets-controller/src/data-sources/RpcDataSource.test.ts @@ -1276,11 +1276,49 @@ describe('RpcDataSource', () => { }); }); - it('updates existing subscription when isUpdate true', async () => { + it('restarts polling on update when the scope changes', async () => { const balanceStartSpy = jest.spyOn( BalanceFetcher.prototype, 'startPolling', ); + const secondAccount = createMockInternalAccount({ + id: 'account-2', + address: '0x9999999999999999999999999999999999999999', + }); + await withController(async ({ controller }) => { + await controller.subscribe({ + request: createDataRequest(), + subscriptionId: 'test-sub', + isUpdate: false, + onAssetsUpdate: jest.fn(), + }); + balanceStartSpy.mockClear(); + + // Adding a second account changes the subscription scope, so polling is + // restarted for the new set of accounts. + await controller.subscribe({ + request: createDataRequest({ + accounts: [createMockInternalAccount(), secondAccount], + }), + subscriptionId: 'test-sub', + isUpdate: true, + onAssetsUpdate: jest.fn(), + }); + expect(balanceStartSpy).toHaveBeenCalledWith( + expect.objectContaining({ accountId: 'account-2' }), + ); + }); + }); + + it('skips restarting polling when re-subscribing with an unchanged scope within one interval', async () => { + const balanceStartSpy = jest.spyOn( + BalanceFetcher.prototype, + 'startPolling', + ); + const balanceStopSpy = jest.spyOn( + BalanceFetcher.prototype, + 'stopPollingByPollingToken', + ); await withController(async ({ controller }) => { await controller.subscribe({ request: createDataRequest(), @@ -1288,13 +1326,46 @@ describe('RpcDataSource', () => { isUpdate: false, onAssetsUpdate: jest.fn(), }); + balanceStartSpy.mockClear(); + balanceStopSpy.mockClear(); + + // Re-subscribing with the identical scope (e.g. a chain flapping back + // from the WebSocket source) is a no-op: polling is neither stopped nor + // restarted, so no redundant immediate RPC fetch is triggered. + await controller.subscribe({ + request: createDataRequest(), + subscriptionId: 'test-sub', + isUpdate: true, + onAssetsUpdate: jest.fn(), + }); + expect(balanceStartSpy).not.toHaveBeenCalled(); + expect(balanceStopSpy).not.toHaveBeenCalled(); + }); + }); + + it('restarts polling when re-subscribing with forceUpdate even if the scope is unchanged', async () => { + const balanceStartSpy = jest.spyOn( + BalanceFetcher.prototype, + 'startPolling', + ); + await withController(async ({ controller }) => { await controller.subscribe({ request: createDataRequest(), subscriptionId: 'test-sub', + isUpdate: false, + onAssetsUpdate: jest.fn(), + }); + balanceStartSpy.mockClear(); + + await controller.subscribe({ + request: createDataRequest({ forceUpdate: true }), + subscriptionId: 'test-sub', isUpdate: true, onAssetsUpdate: jest.fn(), }); - expect(balanceStartSpy).toHaveBeenCalledTimes(2); + expect(balanceStartSpy).toHaveBeenCalledWith( + expect.objectContaining({ accountId: MOCK_ACCOUNT_ID }), + ); }); }); diff --git a/packages/assets-controller/src/data-sources/RpcDataSource.ts b/packages/assets-controller/src/data-sources/RpcDataSource.ts index d28af6e3fd..cd63f841c2 100644 --- a/packages/assets-controller/src/data-sources/RpcDataSource.ts +++ b/packages/assets-controller/src/data-sources/RpcDataSource.ts @@ -35,7 +35,7 @@ import type { DataResponse, Middleware, } from '../types'; -import { normalizeAssetId } from '../utils'; +import { computeSubscriptionScopeKey, normalizeAssetId } from '../utils'; import { ZERO_ADDRESS } from '../utils/constants'; import { AbstractDataSource } from './AbstractDataSource'; import type { @@ -156,6 +156,10 @@ type SubscriptionData = { chains: ChainId[]; /** Accounts being polled */ accounts: InternalAccount[]; + /** Stable scope key for this subscription (accounts + chains + interval + mode). */ + scopeKey: string; + /** Timestamp (ms) when polling for this scope was (re)started. */ + subscribedAt: number; /** Callback to report asset updates to the controller */ onAssetsUpdate: ( response: DataResponse, @@ -1379,17 +1383,45 @@ export class RpcDataSource extends AbstractDataSource< return; } + const pollInterval = + request.updateInterval ?? + this.#balanceFetcher.getIntervalLength() ?? + DEFAULT_BALANCE_INTERVAL; + const scopeKey = computeSubscriptionScopeKey( + request, + chainsToSubscribe, + pollInterval, + ); + + const existing = this.#activeSubscriptions.get(subscriptionId); + + // Skip a redundant restart when an existing subscription already covers the + // exact same scope and its polling (re)started within one interval. + // Restarting would stop/start polling and trigger an immediate duplicate + // RPC fetch for data we just requested (e.g. a chain flapping between the + // WebSocket source and RPC). `forceUpdate` always restarts (it is excluded + // from the scope key). The existing polling keeps refreshing on its own + // interval, so no data is lost. + if ( + !request.forceUpdate && + existing?.scopeKey === scopeKey && + Date.now() - (existing?.subscribedAt ?? 0) < pollInterval + ) { + log('Skipping redundant re-subscribe (scope unchanged)', { + subscriptionId, + chains: chainsToSubscribe, + }); + return; + } + // Handle subscription update - restart polling for new chains - if (isUpdate) { - const existing = this.#activeSubscriptions.get(subscriptionId); - if (existing) { - log('Updating existing subscription - restarting polling', { - subscriptionId, - existingChains: existing.chains, - newChains: chainsToSubscribe, - }); - // Don't return early - continue to unsubscribe and restart polling - } + if (isUpdate && existing) { + log('Updating existing subscription - restarting polling', { + subscriptionId, + existingChains: existing.chains, + newChains: chainsToSubscribe, + }); + // Don't return early - continue to unsubscribe and restart polling } // Clean up existing subscription (stops old polling) @@ -1454,6 +1486,8 @@ export class RpcDataSource extends AbstractDataSource< detectionPollingTokens, chains: chainsToSubscribe, accounts, + scopeKey, + subscribedAt: Date.now(), onAssetsUpdate: subscriptionRequest.onAssetsUpdate, }); diff --git a/packages/assets-controller/src/data-sources/index.ts b/packages/assets-controller/src/data-sources/index.ts index 47cb7c5734..1d8321fcc8 100644 --- a/packages/assets-controller/src/data-sources/index.ts +++ b/packages/assets-controller/src/data-sources/index.ts @@ -21,6 +21,14 @@ export { type BackendWebsocketDataSourceAllowedEvents, } from './BackendWebsocketDataSource'; +export { + AccountActivityDataSource, + createAccountActivityDataSource, + type AccountActivityDataSourceOptions, + type AccountActivityDataSourceState, + type AccountActivityDataSourceAllowedEvents, +} from './AccountActivityDataSource'; + export { RpcDataSource, createRpcDataSource, diff --git a/packages/assets-controller/src/utils/index.ts b/packages/assets-controller/src/utils/index.ts index 408f745261..8d35f30604 100644 --- a/packages/assets-controller/src/utils/index.ts +++ b/packages/assets-controller/src/utils/index.ts @@ -13,3 +13,4 @@ export { buildNativeAssetsFromConstant, buildNativeAssetsFromApi, } from './native-assets'; +export { computeSubscriptionScopeKey } from './subscriptionScope'; diff --git a/packages/assets-controller/src/utils/subscriptionScope.test.ts b/packages/assets-controller/src/utils/subscriptionScope.test.ts new file mode 100644 index 0000000000..476ca8dc0d --- /dev/null +++ b/packages/assets-controller/src/utils/subscriptionScope.test.ts @@ -0,0 +1,124 @@ +import type { InternalAccount } from '@metamask/keyring-internal-api'; + +import type { ChainId, DataRequest } from '../types'; +import { computeSubscriptionScopeKey } from './subscriptionScope'; + +const CHAIN_MAINNET = 'eip155:1' as ChainId; +const CHAIN_POLYGON = 'eip155:137' as ChainId; + +function createAccount(overrides?: Partial): InternalAccount { + return { + id: 'account-1', + address: '0xAbC0000000000000000000000000000000000001', + options: {}, + methods: [], + type: 'eip155:eoa', + scopes: ['eip155:0'], + metadata: { + name: 'Account 1', + keyring: { type: 'HD Key Tree' }, + importTime: 0, + lastSelected: 0, + }, + ...overrides, + } as InternalAccount; +} + +function createRequest(overrides?: Partial): DataRequest { + return { + accountsWithSupportedChains: [ + { account: createAccount(), supportedChains: [CHAIN_MAINNET] }, + ], + chainIds: [CHAIN_MAINNET], + dataTypes: ['balance'], + ...overrides, + }; +} + +describe('computeSubscriptionScopeKey', () => { + it('produces identical keys for equivalent scopes regardless of ordering', () => { + const requestA = createRequest({ + accountsWithSupportedChains: [ + { + account: createAccount({ id: 'a' }), + supportedChains: [CHAIN_MAINNET, CHAIN_POLYGON], + }, + { account: createAccount({ id: 'b' }), supportedChains: [CHAIN_MAINNET] }, + ], + chainIds: [CHAIN_MAINNET, CHAIN_POLYGON], + }); + const requestB = createRequest({ + accountsWithSupportedChains: [ + { account: createAccount({ id: 'b' }), supportedChains: [CHAIN_MAINNET] }, + { + account: createAccount({ id: 'a' }), + supportedChains: [CHAIN_POLYGON, CHAIN_MAINNET], + }, + ], + chainIds: [CHAIN_POLYGON, CHAIN_MAINNET], + }); + + expect( + computeSubscriptionScopeKey(requestA, [CHAIN_MAINNET, CHAIN_POLYGON], 30000), + ).toBe( + computeSubscriptionScopeKey(requestB, [CHAIN_POLYGON, CHAIN_MAINNET], 30000), + ); + }); + + it('is case-insensitive for account addresses', () => { + const lower = createRequest({ + accountsWithSupportedChains: [ + { + account: createAccount({ + address: '0xabc0000000000000000000000000000000000001', + }), + supportedChains: [CHAIN_MAINNET], + }, + ], + }); + const upper = createRequest({ + accountsWithSupportedChains: [ + { + account: createAccount({ + address: '0xABC0000000000000000000000000000000000001', + }), + supportedChains: [CHAIN_MAINNET], + }, + ], + }); + + expect(computeSubscriptionScopeKey(lower, [CHAIN_MAINNET], 30000)).toBe( + computeSubscriptionScopeKey(upper, [CHAIN_MAINNET], 30000), + ); + }); + + it('differs when the chains differ', () => { + const request = createRequest(); + expect(computeSubscriptionScopeKey(request, [CHAIN_MAINNET], 30000)).not.toBe( + computeSubscriptionScopeKey(request, [CHAIN_POLYGON], 30000), + ); + }); + + it('differs when the poll interval differs', () => { + const request = createRequest(); + expect(computeSubscriptionScopeKey(request, [CHAIN_MAINNET], 30000)).not.toBe( + computeSubscriptionScopeKey(request, [CHAIN_MAINNET], 60000), + ); + }); + + it('differs when customAssetsOnly differs', () => { + const regular = createRequest(); + const customOnly = createRequest({ customAssetsOnly: true }); + expect( + computeSubscriptionScopeKey(regular, [CHAIN_MAINNET], 30000), + ).not.toBe(computeSubscriptionScopeKey(customOnly, [CHAIN_MAINNET], 30000)); + }); + + it('ignores forceUpdate so a forced refresh is not treated as a new scope', () => { + const regular = createRequest(); + const forced = createRequest({ forceUpdate: true }); + expect(computeSubscriptionScopeKey(regular, [CHAIN_MAINNET], 30000)).toBe( + computeSubscriptionScopeKey(forced, [CHAIN_MAINNET], 30000), + ); + }); +}); diff --git a/packages/assets-controller/src/utils/subscriptionScope.ts b/packages/assets-controller/src/utils/subscriptionScope.ts new file mode 100644 index 0000000000..5b1843196b --- /dev/null +++ b/packages/assets-controller/src/utils/subscriptionScope.ts @@ -0,0 +1,45 @@ +import type { ChainId, DataRequest } from '../types'; + +/** + * Build a stable, order-independent key describing the effective scope of a + * polling subscription (which accounts, on which chains, at which interval, + * and in which mode). Two subscribe calls that produce the same key would issue + * an identical fetch, so a data source can use this to detect and skip a + * redundant immediate fetch when it is torn down and re-created for the same + * scope (e.g. a chain flapping between the WebSocket source and a polling + * source). + * + * The key intentionally excludes `forceUpdate`: callers that force a refresh + * always want a fresh fetch and should never be treated as redundant. + * + * @param request - The data request for the subscription. + * @param chains - The chains actually being subscribed (may be a subset of + * `request.chainIds`). + * @param pollInterval - The resolved polling interval (ms). + * @returns A deterministic scope key string. + */ +export function computeSubscriptionScopeKey( + request: DataRequest, + chains: ChainId[], + pollInterval: number, +): string { + const accounts = request.accountsWithSupportedChains + .map(({ account, supportedChains }) => { + const scopedChains = [...supportedChains].sort().join(','); + return `${account.id}:${account.address.toLowerCase()}:${scopedChains}`; + }) + .sort() + .join('|'); + + const sortedChains = [...chains].sort().join(','); + const customAssetsOnly = request.customAssetsOnly === true ? '1' : '0'; + const customAssets = [...(request.customAssets ?? [])].sort().join(','); + + return [ + accounts, + sortedChains, + String(pollInterval), + customAssetsOnly, + customAssets, + ].join('#'); +}