feat: Implement BaseDataService (#8039)

## Explanation

This PR implements `BaseDataService` and a function to wrap
`QueryClient` to proxy requests accordingly.

The `BaseDataService`, similarly to the `BaseController` provides the
framework for building a service that can be registered and accessed via
the messenger system, but also provides guarantees about per-request
deduping, retries, caching, invalidation, state-while-revalidate etc via
`@tanstack/query-core`.

The `BaseDataService` provides two utilities for this: `fetchQuery` and
`fetchInfiniteQuery`, which is similar but one is separated for special
pagination behaviour. Each service has its own cache for the APIs that
it exposes that must also be synchronized with the UI processes. To
facilitate this synchronization, the `BaseDataService` also
automatically provides a `cacheUpdate` event.

The overall goal of the PR is to provide a base layer that can keep as
much compatibility as possible with native TanStack Query while also
simultaneously allowing us to have one source of truth per data service.

The synchronization is achieved via a special `QueryClient` created by
`createUIQueryClient`, which wraps functionality such as cache
invalidation, provides the default proxied fetch behaviour and
subscribes to cache updates from data services that it is observing
(e.g. has active queries for).

## References

https://consensyssoftware.atlassian.net/browse/WPC-445

## Checklist

- [x] I've updated the test suite for new or updated code as appropriate
- [x] I've updated documentation (JSDoc, Markdown, etc.) for new or
updated code as appropriate
- [ ] I've communicated my changes to consumers by [updating changelogs
for packages I've
changed](https://github.com/MetaMask/core/tree/main/docs/processes/updating-changelogs.md)
- [ ] I've introduced [breaking
changes](https://github.com/MetaMask/core/tree/main/docs/processes/breaking-changes.md)
in this PR and have prepared draft pull requests for clients and
consumer packages to resolve them

<!-- CURSOR_SUMMARY -->
---

> [!NOTE]
> **Medium Risk**
> Introduces new shared data-service/query infrastructure (caching,
retries/circuit breaking, invalidation, cross-process cache hydration)
that can affect data freshness and event synchronization if misused,
though changes are isolated to new packages with test coverage.
> 
> **Overview**
> Adds an initial implementation of `@metamask/base-data-service`,
introducing `BaseDataService` built on `@tanstack/query-core` with
`fetchQuery`/`fetchInfiniteQuery`, service-policy-wrapped execution
(retries/circuit breaking), a registered `:invalidateQueries` action,
and automatic `:cacheUpdated`/granular `:cacheUpdated:${hash}` events
with dehydrated cache state.
> 
> Adds `@metamask/react-data-query` utilities to consume these services
from UI code: `createUIQueryClient` proxies TanStack queries through a
messenger, subscribes/unsubscribes to per-query cache update events to
hydrate/remove cached entries, and forwards `invalidateQueries` to the
underlying service; also adds typed `useQuery`/`useInfiniteQuery`
wrappers.
> 
> Updates package metadata/build references, adds dependencies (TanStack
v4, controller-utils, messenger, nock, etc.), and adjusts Yarn
constraints to allow the TanStack v4 range and React peer deps without
requiring devDependency installs.
> 
> <sup>Written by [Cursor
Bugbot](https://cursor.com/dashboard?tab=bugbot) for commit
0ae98c7. This will update automatically
on new commits. Configure
[here](https://cursor.com/dashboard?tab=bugbot).</sup>
<!-- /CURSOR_SUMMARY -->

Author: FrederikBolding

packages/base-data-service/CHANGELOG.md

@@ -7,4 +7,8 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0
 
 ## [Unreleased]
 
+### Added
+
+- Initial release ([#8039](https://github.com/MetaMask/core/pull/8039))
+
 [Unreleased]: https://github.com/MetaMask/core/

packages/base-data-service/package.json

@@ -47,12 +47,20 @@
     "test:verbose": "NODE_OPTIONS=--experimental-vm-modules jest --verbose",
     "test:watch": "NODE_OPTIONS=--experimental-vm-modules jest --watch"
   },
+  "dependencies": {
+    "@metamask/controller-utils": "^11.19.0",
+    "@metamask/messenger": "^0.3.0",
+    "@metamask/utils": "^11.9.0",
+    "@tanstack/query-core": "^4.43.0",
+    "fast-deep-equal": "^3.1.3"
+  },
   "devDependencies": {
     "@metamask/auto-changelog": "^3.4.4",
     "@ts-bridge/cli": "^0.6.4",
     "@types/jest": "^29.5.14",
     "deepmerge": "^4.2.2",
     "jest": "^29.7.0",
+    "nock": "^13.3.1",
     "ts-jest": "^29.2.5",
     "typedoc": "^0.25.13",
     "typedoc-plugin-missing-exports": "^2.0.0",

packages/base-data-service/src/BaseDataService.test.ts

@@ -0,0 +1,283 @@
+import { BrokenCircuitError } from '@metamask/controller-utils';
+import { Messenger } from '@metamask/messenger';
+import { hashQueryKey } from '@tanstack/query-core';
+import { cleanAll } from 'nock';
+
+import { ExampleDataService, serviceName } from '../tests/ExampleDataService';
+import {
+  mockAssets,
+  mockTransactionsPage1,
+  mockTransactionsPage2,
+  mockTransactionsPage3,
+  TRANSACTIONS_PAGE_2_CURSOR,
+  TRANSACTIONS_PAGE_3_CURSOR,
+} from '../tests/mocks';
+
+const TEST_ADDRESS = '0x4bbeEB066eD09B7AEd07bF39EEe0460DFa261520';
+
+const MOCK_ASSETS = [
+  'eip155:1/slip44:60',
+  'bip122:000000000019d6689c085ae165831e93/slip44:0',
+  'eip155:1/erc20:0x6b175474e89094c44da98b954eedeac495271d0f',
+];
+
+describe('BaseDataService', () => {
+  beforeEach(() => {
+    mockAssets();
+    mockTransactionsPage1();
+    mockTransactionsPage2();
+    mockTransactionsPage3();
+  });
+
+  it('handles basic queries', async () => {
+    const messenger = new Messenger({ namespace: serviceName });
+    const service = new ExampleDataService(messenger);
+
+    expect(await service.getAssets(MOCK_ASSETS)).toStrictEqual([
+      {
+        assetId: 'eip155:1/erc20:0x6b175474e89094c44da98b954eedeac495271d0f',
+        decimals: 18,
+        name: 'Dai Stablecoin',
+        symbol: 'DAI',
+      },
+      {
+        assetId: 'bip122:000000000019d6689c085ae165831e93/slip44:0',
+        decimals: 8,
+        name: 'Bitcoin',
+        symbol: 'BTC',
+      },
+      {
+        assetId: 'eip155:1/slip44:60',
+        decimals: 18,
+        name: 'Ethereum',
+        symbol: 'ETH',
+      },
+    ]);
+  });
+
+  it('handles paginated queries', async () => {
+    const messenger = new Messenger({ namespace: serviceName });
+    const service = new ExampleDataService(messenger);
+
+    const page1 = await service.getActivity(TEST_ADDRESS);
+
+    expect(page1.data).toHaveLength(3);
+
+    const page2 = await service.getActivity(TEST_ADDRESS, {
+      after: page1.pageInfo.endCursor,
+    });
+
+    expect(page2.data).toHaveLength(3);
+
+    expect(page2.data).not.toStrictEqual(page1.data);
+  });
+
+  it('handles paginated queries starting at a specific page', async () => {
+    const messenger = new Messenger({ namespace: serviceName });
+    const service = new ExampleDataService(messenger);
+
+    const page2 = await service.getActivity(TEST_ADDRESS, {
+      after: TRANSACTIONS_PAGE_2_CURSOR,
+    });
+
+    expect(page2.data).toHaveLength(3);
+
+    const page3 = await service.getActivity(TEST_ADDRESS, {
+      after: page2.pageInfo.endCursor,
+    });
+
+    expect(page3.data).toHaveLength(3);
+
+    expect(page3.data).not.toStrictEqual(page2.data);
+  });
+
+  it('handles backwards queries starting at a specific page', async () => {
+    const messenger = new Messenger({ namespace: serviceName });
+    const service = new ExampleDataService(messenger);
+
+    const page3 = await service.getActivity(TEST_ADDRESS, {
+      after: TRANSACTIONS_PAGE_3_CURSOR,
+    });
+
+    expect(page3.data).toHaveLength(3);
+
+    const page2 = await service.getActivity(TEST_ADDRESS, {
+      before: page3.pageInfo.startCursor,
+    });
+
+    expect(page2.data).toHaveLength(3);
+    expect(page2.data).not.toStrictEqual(page3.data);
+  });
+
+  it('emits `:cacheUpdated` events when cache is updated', async () => {
+    const messenger = new Messenger({ namespace: serviceName });
+    const service = new ExampleDataService(messenger);
+
+    const publishSpy = jest.spyOn(messenger, 'publish');
+
+    await service.getAssets(MOCK_ASSETS);
+
+    const queryKey = ['ExampleDataService:getAssets', MOCK_ASSETS];
+
+    const hash = hashQueryKey(queryKey);
+
+    expect(publishSpy).toHaveBeenNthCalledWith(
+      6,
+      `ExampleDataService:cacheUpdated:${hash}`,
+      {
+        type: 'updated',
+        state: {
+          mutations: [],
+          queries: [
+            expect.objectContaining({
+              state: expect.objectContaining({
+                status: 'success',
+                data: [
+                  {
+                    assetId:
+                      'eip155:1/erc20:0x6b175474e89094c44da98b954eedeac495271d0f',
+                    decimals: 18,
+                    name: 'Dai Stablecoin',
+                    symbol: 'DAI',
+                  },
+                  {
+                    assetId: 'bip122:000000000019d6689c085ae165831e93/slip44:0',
+                    decimals: 8,
+                    name: 'Bitcoin',
+                    symbol: 'BTC',
+                  },
+                  {
+                    assetId: 'eip155:1/slip44:60',
+                    decimals: 18,
+                    name: 'Ethereum',
+                    symbol: 'ETH',
+                  },
+                ],
+              }),
+            }),
+          ],
+        },
+      },
+    );
+  });
+
+  it('emits `:cacheUpdated` events when cache entry is removed', async () => {
+    const messenger = new Messenger({ namespace: serviceName });
+    const service = new ExampleDataService(messenger);
+
+    const publishSpy = jest.spyOn(messenger, 'publish');
+
+    await service.getAssets(MOCK_ASSETS);
+
+    // Wait for GC
+    await new Promise((resolve) => setTimeout(resolve, 0));
+
+    const queryKey = ['ExampleDataService:getAssets', MOCK_ASSETS];
+
+    const hash = hashQueryKey(queryKey);
+
+    expect(publishSpy).toHaveBeenNthCalledWith(
+      8,
+      `ExampleDataService:cacheUpdated:${hash}`,
+      {
+        type: 'removed',
+        state: null,
+      },
+    );
+  });
+
+  it('does not emit events after being destroyed', async () => {
+    const messenger = new Messenger({ namespace: serviceName });
+    const service = new ExampleDataService(messenger);
+    const publishSpy = jest.spyOn(messenger, 'publish');
+
+    service.destroy();
+
+    await service.getAssets(MOCK_ASSETS);
+
+    expect(publishSpy).toHaveBeenCalledTimes(0);
+  });
+
+  it('invalidates queries when requested', async () => {
+    const messenger = new Messenger({ namespace: serviceName });
+    const service = new ExampleDataService(messenger);
+    const publishSpy = jest.spyOn(messenger, 'publish');
+
+    await service.getAssets(MOCK_ASSETS);
+
+    expect(publishSpy).toHaveBeenCalledTimes(6);
+
+    const queryKey = ['ExampleDataService:getAssets', MOCK_ASSETS];
+    await service.invalidateQueries({ queryKey });
+
+    expect(publishSpy).toHaveBeenCalledTimes(8);
+  });
+
+  describe('service policy', () => {
+    beforeEach(() => {
+      cleanAll();
+    });
+
+    it('retries failed queries using the service policy', async () => {
+      const messenger = new Messenger({ namespace: serviceName });
+      const service = new ExampleDataService(messenger);
+
+      mockAssets({ status: 500 });
+      mockAssets({ status: 500 });
+      mockAssets();
+
+      const result = await service.getAssets(MOCK_ASSETS);
+
+      expect(result).toStrictEqual([
+        {
+          assetId: 'eip155:1/erc20:0x6b175474e89094c44da98b954eedeac495271d0f',
+          decimals: 18,
+          name: 'Dai Stablecoin',
+          symbol: 'DAI',
+        },
+        {
+          assetId: 'bip122:000000000019d6689c085ae165831e93/slip44:0',
+          decimals: 8,
+          name: 'Bitcoin',
+          symbol: 'BTC',
+        },
+        {
+          assetId: 'eip155:1/slip44:60',
+          decimals: 18,
+          name: 'Ethereum',
+          symbol: 'ETH',
+        },
+      ]);
+    });
+
+    it('throws after exhausting service policy retries', async () => {
+      const messenger = new Messenger({ namespace: serviceName });
+      const service = new ExampleDataService(messenger);
+
+      mockAssets({ status: 500, body: { error: 'internal server error' } });
+      mockAssets({ status: 500, body: { error: 'internal server error' } });
+      mockAssets({ status: 500, body: { error: 'internal server error' } });
+
+      await expect(service.getAssets(MOCK_ASSETS)).rejects.toThrow(
+        'Query failed with status code: 500.',
+      );
+    });
+
+    it('breaks the circuit after consecutive failures', async () => {
+      const messenger = new Messenger({ namespace: serviceName });
+      const service = new ExampleDataService(messenger);
+
+      mockAssets({ status: 500, body: { error: 'internal server error' } });
+      mockAssets({ status: 500, body: { error: 'internal server error' } });
+      mockAssets({ status: 500, body: { error: 'internal server error' } });
+
+      await expect(service.getAssets(MOCK_ASSETS)).rejects.toThrow(
+        'Query failed with status code: 500.',
+      );
+
+      await expect(service.getAssets(MOCK_ASSETS)).rejects.toThrow(
+        BrokenCircuitError,
+      );
+    });
+  });
+});