diff --git a/.github/workflows/build-lint-test.yml b/.github/workflows/build-lint-test.yml index ea8355988..63db0c98a 100644 --- a/.github/workflows/build-lint-test.yml +++ b/.github/workflows/build-lint-test.yml @@ -11,6 +11,14 @@ on: required: true RPC_URL_LOCALNET_LIST: required: true + RPC_WEB_SOCKET_URL_MAINNET: + required: true + RPC_WEB_SOCKET_URL_DEVNET: + required: true + RPC_WEB_SOCKET_URL_TESTNET: + required: true + RPC_WEB_SOCKET_URL_LOCALNET: + required: true PRICE_API_BASE_URL: required: true LOCAL_API_BASE_URL: @@ -47,6 +55,10 @@ jobs: RPC_URL_DEVNET_LIST: ${{ secrets.RPC_URL_DEVNET_LIST }} RPC_URL_TESTNET_LIST: ${{ secrets.RPC_URL_TESTNET_LIST }} RPC_URL_LOCALNET_LIST: ${{ secrets.RPC_URL_LOCALNET_LIST }} + RPC_WEB_SOCKET_URL_MAINNET: ${{ secrets.RPC_WEB_SOCKET_URL_MAINNET }} + RPC_WEB_SOCKET_URL_DEVNET: ${{ secrets.RPC_WEB_SOCKET_URL_DEVNET }} + RPC_WEB_SOCKET_URL_TESTNET: ${{ secrets.RPC_WEB_SOCKET_URL_TESTNET }} + RPC_WEB_SOCKET_URL_LOCALNET: ${{ secrets.RPC_WEB_SOCKET_URL_LOCALNET }} EXPLORER_BASE_URL: ${{ secrets.EXPLORER_BASE_URL }} PRICE_API_BASE_URL: ${{ secrets.PRICE_API_BASE_URL }} LOCAL_API_BASE_URL: ${{ secrets.LOCAL_API_BASE_URL }} @@ -94,6 +106,10 @@ jobs: RPC_URL_DEVNET_LIST: ${{ secrets.RPC_URL_DEVNET_LIST }} RPC_URL_TESTNET_LIST: ${{ secrets.RPC_URL_TESTNET_LIST }} RPC_URL_LOCALNET_LIST: ${{ secrets.RPC_URL_LOCALNET_LIST }} + RPC_WEB_SOCKET_URL_MAINNET: ${{ secrets.RPC_WEB_SOCKET_URL_MAINNET }} + RPC_WEB_SOCKET_URL_DEVNET: ${{ secrets.RPC_WEB_SOCKET_URL_DEVNET }} + RPC_WEB_SOCKET_URL_TESTNET: ${{ secrets.RPC_WEB_SOCKET_URL_TESTNET }} + RPC_WEB_SOCKET_URL_LOCALNET: ${{ secrets.RPC_WEB_SOCKET_URL_LOCALNET }} EXPLORER_BASE_URL: ${{ secrets.EXPLORER_BASE_URL }} PRICE_API_BASE_URL: ${{ secrets.PRICE_API_BASE_URL }} LOCAL_API_BASE_URL: ${{ secrets.LOCAL_API_BASE_URL }} diff --git a/.github/workflows/create-release-pr.yml b/.github/workflows/create-release-pr.yml index 8cfb9a9f0..1d7d82e55 100644 --- a/.github/workflows/create-release-pr.yml +++ b/.github/workflows/create-release-pr.yml @@ -59,6 +59,10 @@ jobs: RPC_URL_DEVNET_LIST: ${{ secrets.RPC_URL_DEVNET_LIST }} RPC_URL_TESTNET_LIST: ${{ secrets.RPC_URL_TESTNET_LIST }} RPC_URL_LOCALNET_LIST: ${{ secrets.RPC_URL_LOCALNET_LIST }} + RPC_WEB_SOCKET_URL_MAINNET: ${{ secrets.RPC_WEB_SOCKET_URL_MAINNET }} + RPC_WEB_SOCKET_URL_DEVNET: ${{ secrets.RPC_WEB_SOCKET_URL_DEVNET }} + RPC_WEB_SOCKET_URL_TESTNET: ${{ secrets.RPC_WEB_SOCKET_URL_TESTNET }} + RPC_WEB_SOCKET_URL_LOCALNET: ${{ secrets.RPC_WEB_SOCKET_URL_LOCALNET }} EXPLORER_BASE_URL: ${{ secrets.EXPLORER_BASE_URL }} PRICE_API_BASE_URL: ${{ secrets.PRICE_API_BASE_URL }} LOCAL_API_BASE_URL: ${{ secrets.LOCAL_API_BASE_URL }} diff --git a/.github/workflows/main.yml b/.github/workflows/main.yml index ded273a8b..171f98bc3 100644 --- a/.github/workflows/main.yml +++ b/.github/workflows/main.yml @@ -13,6 +13,10 @@ jobs: RPC_URL_DEVNET_LIST: ${{ secrets.RPC_URL_DEVNET_LIST }} RPC_URL_TESTNET_LIST: ${{ secrets.RPC_URL_TESTNET_LIST }} RPC_URL_LOCALNET_LIST: ${{ secrets.RPC_URL_LOCALNET_LIST }} + RPC_WEB_SOCKET_URL_MAINNET: ${{ secrets.RPC_WEB_SOCKET_URL_MAINNET }} + RPC_WEB_SOCKET_URL_DEVNET: ${{ secrets.RPC_WEB_SOCKET_URL_DEVNET }} + RPC_WEB_SOCKET_URL_TESTNET: ${{ secrets.RPC_WEB_SOCKET_URL_TESTNET }} + RPC_WEB_SOCKET_URL_LOCALNET: ${{ secrets.RPC_WEB_SOCKET_URL_LOCALNET }} EXPLORER_BASE_URL: ${{ secrets.EXPLORER_BASE_URL }} PRICE_API_BASE_URL: ${{ secrets.PRICE_API_BASE_URL }} LOCAL_API_BASE_URL: ${{ secrets.LOCAL_API_BASE_URL }} @@ -87,6 +91,10 @@ jobs: RPC_URL_DEVNET_LIST: ${{ secrets.RPC_URL_DEVNET_LIST }} RPC_URL_TESTNET_LIST: ${{ secrets.RPC_URL_TESTNET_LIST }} RPC_URL_LOCALNET_LIST: ${{ secrets.RPC_URL_LOCALNET_LIST }} + RPC_WEB_SOCKET_URL_MAINNET: ${{ secrets.RPC_WEB_SOCKET_URL_MAINNET }} + RPC_WEB_SOCKET_URL_DEVNET: ${{ secrets.RPC_WEB_SOCKET_URL_DEVNET }} + RPC_WEB_SOCKET_URL_TESTNET: ${{ secrets.RPC_WEB_SOCKET_URL_TESTNET }} + RPC_WEB_SOCKET_URL_LOCALNET: ${{ secrets.RPC_WEB_SOCKET_URL_LOCALNET }} EXPLORER_BASE_URL: ${{ secrets.EXPLORER_BASE_URL }} PRICE_API_BASE_URL: ${{ secrets.PRICE_API_BASE_URL }} LOCAL_API_BASE_URL: ${{ secrets.LOCAL_API_BASE_URL }} diff --git a/.github/workflows/pr.yml b/.github/workflows/pr.yml index f95423b0e..cd61c2a63 100644 --- a/.github/workflows/pr.yml +++ b/.github/workflows/pr.yml @@ -27,6 +27,10 @@ jobs: RPC_URL_DEVNET_LIST: ${{ secrets.RPC_URL_DEVNET_LIST }} RPC_URL_TESTNET_LIST: ${{ secrets.RPC_URL_TESTNET_LIST }} RPC_URL_LOCALNET_LIST: ${{ secrets.RPC_URL_LOCALNET_LIST }} + RPC_WEB_SOCKET_URL_MAINNET: ${{ secrets.RPC_WEB_SOCKET_URL_MAINNET }} + RPC_WEB_SOCKET_URL_DEVNET: ${{ secrets.RPC_WEB_SOCKET_URL_DEVNET }} + RPC_WEB_SOCKET_URL_TESTNET: ${{ secrets.RPC_WEB_SOCKET_URL_TESTNET }} + RPC_WEB_SOCKET_URL_LOCALNET: ${{ secrets.RPC_WEB_SOCKET_URL_LOCALNET }} EXPLORER_BASE_URL: ${{ secrets.EXPLORER_BASE_URL }} PRICE_API_BASE_URL: ${{ secrets.PRICE_API_BASE_URL }} LOCAL_API_BASE_URL: ${{ secrets.LOCAL_API_BASE_URL }} diff --git a/.github/workflows/publish-release.yml b/.github/workflows/publish-release.yml index 50ec48da8..c4decd447 100644 --- a/.github/workflows/publish-release.yml +++ b/.github/workflows/publish-release.yml @@ -15,6 +15,14 @@ on: required: true RPC_URL_LOCALNET_LIST: required: true + RPC_WEB_SOCKET_URL_MAINNET: + required: true + RPC_WEB_SOCKET_URL_DEVNET: + required: true + RPC_WEB_SOCKET_URL_TESTNET: + required: true + RPC_WEB_SOCKET_URL_LOCALNET: + required: true PRICE_API_BASE_URL: required: true LOCAL_API_BASE_URL: @@ -51,6 +59,10 @@ jobs: RPC_URL_DEVNET_LIST: ${{ secrets.RPC_URL_DEVNET_LIST }} RPC_URL_TESTNET_LIST: ${{ secrets.RPC_URL_TESTNET_LIST }} RPC_URL_LOCALNET_LIST: ${{ secrets.RPC_URL_LOCALNET_LIST }} + RPC_WEB_SOCKET_URL_MAINNET: ${{ secrets.RPC_WEB_SOCKET_URL_MAINNET }} + RPC_WEB_SOCKET_URL_DEVNET: ${{ secrets.RPC_WEB_SOCKET_URL_DEVNET }} + RPC_WEB_SOCKET_URL_TESTNET: ${{ secrets.RPC_WEB_SOCKET_URL_TESTNET }} + RPC_WEB_SOCKET_URL_LOCALNET: ${{ secrets.RPC_WEB_SOCKET_URL_LOCALNET }} EXPLORER_BASE_URL: ${{ secrets.EXPLORER_BASE_URL }} PRICE_API_BASE_URL: ${{ secrets.PRICE_API_BASE_URL }} LOCAL_API_BASE_URL: ${{ secrets.LOCAL_API_BASE_URL }} diff --git a/packages/site/src/components/Handlers/Handlers.tsx b/packages/site/src/components/Handlers/Handlers.tsx index c7f457afd..e7f40f5cb 100644 --- a/packages/site/src/components/Handlers/Handlers.tsx +++ b/packages/site/src/components/Handlers/Handlers.tsx @@ -1,6 +1,7 @@ import { Text as ChakraText, Flex, Stack } from '@chakra-ui/react'; import { OnProtocolRequest } from './OnProtocolRequest'; +import { WebSockets } from './WebSockets'; export const Handlers = () => ( @@ -9,6 +10,7 @@ export const Handlers = () => ( + ); diff --git a/packages/site/src/components/Handlers/WebSockets.tsx b/packages/site/src/components/Handlers/WebSockets.tsx new file mode 100644 index 000000000..1bd9026bf --- /dev/null +++ b/packages/site/src/components/Handlers/WebSockets.tsx @@ -0,0 +1,68 @@ +/* eslint-disable @typescript-eslint/no-unused-vars */ +import { Button, Card, Flex } from '@chakra-ui/react'; + +import { TestDappRpcRequestMethod } from '../../../../snap/src/core/handlers/onRpcRequest/types'; +import { useInvokeSnap } from '../../hooks'; +import { useShowToasterForResponse } from '../../hooks/useToasterForResponse'; + +export const WebSockets = () => { + const invokeSnap = useInvokeSnap(); + const { showToasterForResponse } = useShowToasterForResponse(); + + const setupAllConnections = async () => { + showToasterForResponse( + { result: 'ok' }, + { + title: 'Initiated WebSocket connections to all active networks', + description: `Inspect the offscreen console`, + }, + ); + + await invokeSnap({ + method: TestDappRpcRequestMethod.TestSetupAllConnections, + }); + }; + + const subscribeToAccount = async () => { + await invokeSnap({ + method: TestDappRpcRequestMethod.TestSubscribeToAccount, + }); + + showToasterForResponse( + { result: 'ok' }, + { + title: + 'Initiated subscription to account 8A4AptCThfbuknsbteHgGKXczfJpfjuVA9SLTSGaaLGC', + description: `Inspect the offscreen console`, + }, + ); + }; + + return ( + + + WebSockets + + + + + + + + + + + ); +}; diff --git a/packages/snap/.env.sample b/packages/snap/.env.sample index b371567ea..325ab4823 100644 --- a/packages/snap/.env.sample +++ b/packages/snap/.env.sample @@ -10,6 +10,12 @@ RPC_URL_MAINNET_LIST= RPC_URL_DEVNET_LIST= RPC_URL_TESTNET_LIST= RPC_URL_LOCALNET_LIST= + +RPC_WEB_SOCKET_URL_MAINNET= +RPC_WEB_SOCKET_URL_DEVNET= +RPC_WEB_SOCKET_URL_TESTNET= +RPC_WEB_SOCKET_URL_LOCALNET= + EXPLORER_BASE_URL= PRICE_API_BASE_URL= diff --git a/packages/snap/snap.config.ts b/packages/snap/snap.config.ts index 74485b1ce..dcbb25e0c 100644 --- a/packages/snap/snap.config.ts +++ b/packages/snap/snap.config.ts @@ -11,6 +11,10 @@ const environment = { RPC_URL_DEVNET_LIST: process.env.RPC_URL_DEVNET_LIST ?? '', RPC_URL_TESTNET_LIST: process.env.RPC_URL_TESTNET_LIST ?? '', RPC_URL_LOCALNET_LIST: process.env.RPC_URL_LOCALNET_LIST ?? '', + RPC_WEB_SOCKET_URL_MAINNET: process.env.RPC_WEB_SOCKET_URL_MAINNET ?? '', + RPC_WEB_SOCKET_URL_DEVNET: process.env.RPC_WEB_SOCKET_URL_DEVNET ?? '', + RPC_WEB_SOCKET_URL_TESTNET: process.env.RPC_WEB_SOCKET_URL_TESTNET ?? '', + RPC_WEB_SOCKET_URL_LOCALNET: process.env.RPC_WEB_SOCKET_URL_LOCALNET ?? '', EXPLORER_BASE_URL: process.env.EXPLORER_BASE_URL ?? '', PRICE_API_BASE_URL: process.env.PRICE_API_BASE_URL ?? '', TOKEN_API_BASE_URL: process.env.TOKEN_API_BASE_URL ?? '', diff --git a/packages/snap/snap.manifest.json b/packages/snap/snap.manifest.json index c41db9591..22a0322c9 100644 --- a/packages/snap/snap.manifest.json +++ b/packages/snap/snap.manifest.json @@ -7,7 +7,7 @@ "url": "https://github.com/MetaMask/snap-solana-wallet.git" }, "source": { - "shasum": "U6qs3tA3phhscX54m+BAD5Mi1aBgT9p9g3qQ2qHrCsQ=", + "shasum": "XSE/gfpj/gSu5aA9IihnjJ+IczHvXAdJN1Z/EegXk7w=", "location": { "npm": { "filePath": "dist/bundle.js", diff --git a/packages/snap/src/core/clients/price-api/PriceApiClient.test.ts b/packages/snap/src/core/clients/price-api/PriceApiClient.test.ts index 71acfef3f..1c384be8e 100644 --- a/packages/snap/src/core/clients/price-api/PriceApiClient.test.ts +++ b/packages/snap/src/core/clients/price-api/PriceApiClient.test.ts @@ -264,7 +264,9 @@ describe('PriceApiClient', () => { mockFetch, mockLogger, ), - ).toThrow('URL must use http or https protocol'); + ).toThrow( + 'URL must use one of the following protocols: http:,https:,wss:', + ); }); it('rejects tokenCaip19Ids that are invalid or that include malicious inputs', async () => { diff --git a/packages/snap/src/core/handlers/index.ts b/packages/snap/src/core/handlers/index.ts new file mode 100644 index 000000000..fe5130d9d --- /dev/null +++ b/packages/snap/src/core/handlers/index.ts @@ -0,0 +1 @@ +export * from './onClientRequest'; diff --git a/packages/snap/src/core/handlers/onCronjob/cronjobs/refreshSend.tsx b/packages/snap/src/core/handlers/onCronjob/cronjobs/refreshSend.tsx index f73f2a83e..86f105cf0 100644 --- a/packages/snap/src/core/handlers/onCronjob/cronjobs/refreshSend.tsx +++ b/packages/snap/src/core/handlers/onCronjob/cronjobs/refreshSend.tsx @@ -1,10 +1,10 @@ import type { CaipAssetType, OnCronjobHandler } from '@metamask/snaps-sdk'; -import { type SpotPrices } from 'src/core/clients/price-api/types'; import { DEFAULT_SEND_CONTEXT } from '../../../../features/send/render'; import { Send } from '../../../../features/send/Send'; import type { SendContext } from '../../../../features/send/types'; import { priceApiClient, state } from '../../../../snapContext'; +import type { SpotPrices } from '../../../clients/price-api/types'; import type { UnencryptedStateValue } from '../../../services/state/State'; import { getInterfaceContextOrThrow, diff --git a/packages/snap/src/core/handlers/onKeyringRequest/Keyring.test.ts b/packages/snap/src/core/handlers/onKeyringRequest/Keyring.test.ts index 95a7db24a..543c15165 100644 --- a/packages/snap/src/core/handlers/onKeyringRequest/Keyring.test.ts +++ b/packages/snap/src/core/handlers/onKeyringRequest/Keyring.test.ts @@ -219,6 +219,7 @@ describe('SolanaKeyring', () => { metadata: {}, tokenPrices: {}, signatures: {}, + subscriptions: {}, }); // Start with no accounts keyring = new SolanaKeyring({ diff --git a/packages/snap/src/core/handlers/onKeyringRequest/Keyring.ts b/packages/snap/src/core/handlers/onKeyringRequest/Keyring.ts index 2a3fede7b..8be633a55 100644 --- a/packages/snap/src/core/handlers/onKeyringRequest/Keyring.ts +++ b/packages/snap/src/core/handlers/onKeyringRequest/Keyring.ts @@ -38,7 +38,7 @@ import { asStrictKeyringAccount, type SolanaKeyringAccount, } from '../../../entities'; -import { type Network } from '../../constants/solana'; +import type { Network } from '../../constants/solana'; import type { AssetsService } from '../../services/assets/AssetsService'; import type { ConfirmationHandler } from '../../services/confirmation/ConfirmationHandler'; import type { IStateManager } from '../../services/state/IStateManager'; diff --git a/packages/snap/src/core/handlers/onRpcRequest/index.ts b/packages/snap/src/core/handlers/onRpcRequest/index.ts index 00b380cfd..cc14c2f8d 100644 --- a/packages/snap/src/core/handlers/onRpcRequest/index.ts +++ b/packages/snap/src/core/handlers/onRpcRequest/index.ts @@ -1,10 +1,21 @@ import { type OnRpcRequestHandler } from '@metamask/snaps-sdk'; import { renderSend } from '../../../features/send/render'; +import { eventEmitter } from '../../../snapContext'; import { getFeeForTransaction } from './getFeeForTransaction'; -import { RpcRequestMethod } from './types'; +import { RpcRequestMethod, TestDappRpcRequestMethod } from './types'; export const handlers: Record = { [RpcRequestMethod.StartSendTransactionFlow]: renderSend, [RpcRequestMethod.GetFeeForTransaction]: getFeeForTransaction, + + // Methods specific to the test dapp + [TestDappRpcRequestMethod.TestSetupAllConnections as any]: async () => { + await eventEmitter.emitSync('onTestSetupAllConnections'); + return null; + }, + [TestDappRpcRequestMethod.TestSubscribeToAccount as any]: async () => { + await eventEmitter.emitSync('onTestSubscribeToAccount'); + return null; + }, }; diff --git a/packages/snap/src/core/handlers/onRpcRequest/types.ts b/packages/snap/src/core/handlers/onRpcRequest/types.ts index 083f1fa46..580464809 100644 --- a/packages/snap/src/core/handlers/onRpcRequest/types.ts +++ b/packages/snap/src/core/handlers/onRpcRequest/types.ts @@ -2,3 +2,12 @@ export enum RpcRequestMethod { StartSendTransactionFlow = 'startSendTransactionFlow', GetFeeForTransaction = 'getFeeForTransaction', } + +/** + * Methods specific to the test dapp, + * to allow specific flows for manual testing. + */ +export enum TestDappRpcRequestMethod { + TestSetupAllConnections = 'testSetupAllConnections', + TestSubscribeToAccount = 'testSubscribeToAccount', +} diff --git a/packages/snap/src/core/ports/index.ts b/packages/snap/src/core/ports/index.ts new file mode 100644 index 000000000..1477200e3 --- /dev/null +++ b/packages/snap/src/core/ports/index.ts @@ -0,0 +1 @@ +export * from './subscriptions'; diff --git a/packages/snap/src/core/ports/subscriptions/ConnectionManagerPort.ts b/packages/snap/src/core/ports/subscriptions/ConnectionManagerPort.ts new file mode 100644 index 000000000..0c88aa2cb --- /dev/null +++ b/packages/snap/src/core/ports/subscriptions/ConnectionManagerPort.ts @@ -0,0 +1,30 @@ +import type { Network } from '../../constants/solana'; + +/** + * Port interface for managing the lifecycle of JSON-RPC subscription connections. + * It is responsible for opening and closing connections for each network, + * and for re-opening connections when they are dropped. + */ +export type ConnectionManagerPort = { + /** + * Sets up connections for all networks. + * - Opens the connections for all enabled networks that are not already open. + * - Closes the connections for all disabled networks. + * @returns A promise that resolves when the connections are setup. + */ + setupAllConnections(): Promise; + + /** + * Registers a callback to be called when connection is recovered. + * @param network - The network to register the callback for. + * @param callback - The callback function to register. + */ + onConnectionRecovery(network: Network, callback: () => Promise): void; + + /** + * Gets the connection ID for the specified network. + * @param network - The network to get the connection ID for. + * @returns The connection ID, or null if no connection exists for the network. + */ + getConnectionIdByNetwork(network: Network): Promise; +}; diff --git a/packages/snap/src/core/ports/subscriptions/SubscriberPort.ts b/packages/snap/src/core/ports/subscriptions/SubscriberPort.ts new file mode 100644 index 000000000..b189419e6 --- /dev/null +++ b/packages/snap/src/core/ports/subscriptions/SubscriberPort.ts @@ -0,0 +1,65 @@ +import type { SubscriptionRequest } from '../../../entities'; + +export type SubscriptionCallbacks = { + /** + * A callback that will be called when a notification is received. + * For instance, if the subscription is for the `accountSubscribe` method, the callback will be called every time the account changes. + */ + onNotification: (message: any) => Promise; + /** + * A callback that will be called when the subscription fails. + */ + onSubscriptionFailed?: (error: any) => Promise; + /** + * A callback that will be called: + * - when the connection is opened (in case subscription was requested before it was opened). + * - when the connection is re-opened after it was lost. + * + * This is the "message gap recovery", that allows us to compensate for potential missed messages. + * When a connection is lost unexpectedly, any messages we miss while disconnected can result in the UI falling behind or becoming corrupt. + * This callback should typically contain an HTTP fetch to catch-up with the latest state. + * + * @example + * ```ts + * // Connection is open... + * + * const subscriber = new SubscriberAdapter(...); + * await subscriber.subscribe({method: "accountSubscribe", ...}, { + * onConnectionRecovery: async () => { + * // Fetch the latest account state from the server. + * const account = await fetchAccount(accountAddress); + * // Update the state + * }, + * }); + * + * // Connection is lost... + * // Some changes happen on the account, that are missed while disconnected. + * // Connection is re-opened... + * // The onConnectionRecovery is called, ensuring we update the state with the latest account state. + * // Subscription is re-established, and we receive again the future changes. + * ``` + */ + onConnectionRecovery?: () => Promise; +}; + +/** + * A port for subscribing / unsubscribing to JSON-RPC subscriptions. + */ +export type SubscriberPort = { + /** + * Subscribes to a JSON-RPC subscription. + * @param request - The subscription request. + * @param onNotification - The callback to be called when a notification is received. + * @returns The subscription ID. + */ + subscribe( + request: SubscriptionRequest, + callbacks: SubscriptionCallbacks, + ): Promise; + + /** + * Unsubscribes from a JSON-RPC subscription. + * @param subscriptionId - The subscription ID. + */ + unsubscribe(subscriptionId: string): Promise; +}; diff --git a/packages/snap/src/core/ports/subscriptions/index.ts b/packages/snap/src/core/ports/subscriptions/index.ts new file mode 100644 index 000000000..7c97f1c80 --- /dev/null +++ b/packages/snap/src/core/ports/subscriptions/index.ts @@ -0,0 +1,2 @@ +export * from './ConnectionManagerPort'; +export * from './SubscriberPort'; diff --git a/packages/snap/src/core/services/config/ConfigProvider.ts b/packages/snap/src/core/services/config/ConfigProvider.ts index 9bbf43606..9a9a0d536 100644 --- a/packages/snap/src/core/services/config/ConfigProvider.ts +++ b/packages/snap/src/core/services/config/ConfigProvider.ts @@ -36,6 +36,10 @@ const EnvStruct = object({ RPC_URL_DEVNET_LIST: CommaSeparatedListOfUrlsStruct, RPC_URL_TESTNET_LIST: CommaSeparatedListOfUrlsStruct, RPC_URL_LOCALNET_LIST: CommaSeparatedListOfStringsStruct, + RPC_WEB_SOCKET_URL_MAINNET: UrlStruct, + RPC_WEB_SOCKET_URL_DEVNET: UrlStruct, + RPC_WEB_SOCKET_URL_TESTNET: UrlStruct, + RPC_WEB_SOCKET_URL_LOCALNET: UrlStruct, EXPLORER_BASE_URL: UrlStruct, PRICE_API_BASE_URL: UrlStruct, TOKEN_API_BASE_URL: UrlStruct, @@ -48,6 +52,7 @@ export type Env = Infer; export type NetworkWithRpcUrls = (typeof Networks)[Network] & { rpcUrls: string[]; + webSocketUrl: string; }; export type Config = { @@ -72,6 +77,10 @@ export type Config = { securityAlertsApi: { baseUrl: string; }; + subscription: { + maxReconnectAttempts: number; + reconnectDelayMilliseconds: number; + }; }; /** @@ -99,16 +108,15 @@ export class ConfigProvider { RPC_URL_DEVNET_LIST: process.env.RPC_URL_DEVNET_LIST, RPC_URL_TESTNET_LIST: process.env.RPC_URL_TESTNET_LIST, RPC_URL_LOCALNET_LIST: process.env.RPC_URL_LOCALNET_LIST, + RPC_WEB_SOCKET_URL_MAINNET: process.env.RPC_WEB_SOCKET_URL_MAINNET, + RPC_WEB_SOCKET_URL_DEVNET: process.env.RPC_WEB_SOCKET_URL_DEVNET, + RPC_WEB_SOCKET_URL_TESTNET: process.env.RPC_WEB_SOCKET_URL_TESTNET, + RPC_WEB_SOCKET_URL_LOCALNET: process.env.RPC_WEB_SOCKET_URL_LOCALNET, EXPLORER_BASE_URL: process.env.EXPLORER_BASE_URL, - // Price API PRICE_API_BASE_URL: process.env.PRICE_API_BASE_URL, - // Token API TOKEN_API_BASE_URL: process.env.TOKEN_API_BASE_URL, - // Static API STATIC_API_BASE_URL: process.env.STATIC_API_BASE_URL, - // Blockaid - SECURITY_ALERTS_API_BASE_URL: process.env.SECURITY_ALERTS_API_BASE_URL, - // Local API + SECURITY_ALERTS_API_BASE_URL: process.env.SECURITY_ALERTS_API_BASE_URL, // Blockaid LOCAL_API_BASE_URL: process.env.LOCAL_API_BASE_URL, }; @@ -123,18 +131,22 @@ export class ConfigProvider { { ...Networks[Network.Mainnet], rpcUrls: environment.RPC_URL_MAINNET_LIST, + webSocketUrl: environment.RPC_WEB_SOCKET_URL_MAINNET, }, { ...Networks[Network.Devnet], rpcUrls: environment.RPC_URL_DEVNET_LIST, + webSocketUrl: environment.RPC_WEB_SOCKET_URL_DEVNET, }, { ...Networks[Network.Testnet], rpcUrls: environment.RPC_URL_TESTNET_LIST, + webSocketUrl: environment.RPC_WEB_SOCKET_URL_TESTNET, }, { ...Networks[Network.Localnet], rpcUrls: environment.RPC_URL_LOCALNET_LIST, + webSocketUrl: environment.RPC_WEB_SOCKET_URL_LOCALNET, }, ], explorerBaseUrl: environment.EXPLORER_BASE_URL, @@ -165,6 +177,10 @@ export class ConfigProvider { ? environment.LOCAL_API_BASE_URL : environment.SECURITY_ALERTS_API_BASE_URL, }, + subscription: { + maxReconnectAttempts: 5, + reconnectDelayMilliseconds: 1000, + }, }; } diff --git a/packages/snap/src/core/services/state/IStateManager.ts b/packages/snap/src/core/services/state/IStateManager.ts index 5060a0b14..48e258b99 100644 --- a/packages/snap/src/core/services/state/IStateManager.ts +++ b/packages/snap/src/core/services/state/IStateManager.ts @@ -5,7 +5,7 @@ export type IStateManager> = { * Gets the whole state object. * * ⚠️ WARNING: Use with caution because it transfers the whole state, which might contain a lot of data. - * If you need to retrieve only a specific part of the state, use {@link IStateManager.getKey} instead. + * If you need to retrieve only a specific part of the state, use IStateManager.getKey instead. * * @example * ```typescript diff --git a/packages/snap/src/core/services/state/State.ts b/packages/snap/src/core/services/state/State.ts index bdb4a9fd0..a28c36c0a 100644 --- a/packages/snap/src/core/services/state/State.ts +++ b/packages/snap/src/core/services/state/State.ts @@ -5,7 +5,7 @@ import type { CaipAssetType } from '@metamask/utils'; import type { Address, Signature } from '@solana/kit'; import { unset } from 'lodash'; -import type { SolanaKeyringAccount } from '../../../entities'; +import type { SolanaKeyringAccount, Subscription } from '../../../entities'; import type { SpotPrices } from '../../clients/price-api/types'; import { deserialize } from '../../serialization/deserialize'; import { serialize } from '../../serialization/serialize'; @@ -24,6 +24,7 @@ export type UnencryptedStateValue = { signatures: Record; assets: Record>; tokenPrices: SpotPrices; + subscriptions: Record; }; export const DEFAULT_UNENCRYPTED_STATE: UnencryptedStateValue = { @@ -33,6 +34,7 @@ export const DEFAULT_UNENCRYPTED_STATE: UnencryptedStateValue = { signatures: {}, assets: {}, tokenPrices: {}, + subscriptions: {}, }; export type StateConfig> = { diff --git a/packages/snap/src/core/validation/structs.test.ts b/packages/snap/src/core/validation/structs.test.ts index 0121a300d..ad1eed6dd 100644 --- a/packages/snap/src/core/validation/structs.test.ts +++ b/packages/snap/src/core/validation/structs.test.ts @@ -33,6 +33,7 @@ describe('structs', () => { 'https://example.com/path-with-hyphens', 'https://example.com/path_with_underscore', 'http://localhost:8899', + 'wss://example.com', ]; validUrls.forEach((url) => { diff --git a/packages/snap/src/core/validation/structs.ts b/packages/snap/src/core/validation/structs.ts index d2d34e56e..65aa8843f 100644 --- a/packages/snap/src/core/validation/structs.ts +++ b/packages/snap/src/core/validation/structs.ts @@ -1,5 +1,5 @@ import { CaipAssetTypeStruct, SolMethod } from '@metamask/keyring-api'; -import type { Infer, Struct } from '@metamask/superstruct'; +import type { Struct } from '@metamask/superstruct'; import { array, define, @@ -28,10 +28,10 @@ export const PositiveNumberStringStruct = pattern( ); /** - * Validates that a string is a valid and safe URL. Accepts http and https protocols. + * Validates that a string is a valid and safe URL. * * It rejects: - * - Non-HTTP/HTTPS protocols + * - Non-HTTP/HTTPS/WSS protocols * - Malformed URL format or incorrect protocol format * - Invalid hostname format (must follow proper domain naming conventions) * - Protocol pollution attempts (backslashes, @ symbol, %2f@, %5c@) @@ -56,12 +56,13 @@ export const UrlStruct = refine(string(), 'safe-url', (value) => { const url = new URL(value); // Protocol check - if (!['http:', 'https:'].includes(url.protocol)) { - return 'URL must use http or https protocol'; + const supportedProtocols = ['http:', 'https:', 'wss:']; + if (!supportedProtocols.includes(url.protocol)) { + return `URL must use one of the following protocols: ${supportedProtocols}`; } // Validate URL format - if (!value.match(/^https?:\/\/[^/]+\/?/u)) { + if (!value.match(/^(https?|wss):\/\/[^/]+\/?/u)) { return 'Malformed URL - incorrect protocol format'; } diff --git a/packages/snap/src/entities/index.ts b/packages/snap/src/entities/index.ts index 3f0d7be4f..952da4491 100644 --- a/packages/snap/src/entities/index.ts +++ b/packages/snap/src/entities/index.ts @@ -1,2 +1,3 @@ export * from './instructions'; export * from './keyring-account'; +export * from './subscriptions'; diff --git a/packages/snap/src/entities/subscriptions.ts b/packages/snap/src/entities/subscriptions.ts new file mode 100644 index 000000000..088716b25 --- /dev/null +++ b/packages/snap/src/entities/subscriptions.ts @@ -0,0 +1,37 @@ +import type { GetWebSocketsResult } from '@metamask/snaps-sdk'; +import type { JsonRpcParams } from '@metamask/utils'; + +import type { Network } from '../core/constants/solana'; + +/** + * A request to subscribe to a JSON-RPC subscription. + */ +export type SubscriptionRequest = { + method: string; + unsubscribeMethod: string; + params: JsonRpcParams; + network: Network; +}; + +export type Connection = GetWebSocketsResult[number]; + +/** + * Once the Subscriber acknowledges the subscription request, + * it generates a subscrption ID, and the subscription is pending (waiting for the confirmation message). + */ +export type PendingSubscription = SubscriptionRequest & { + readonly id: string; + readonly status: 'pending'; + readonly requestId: string; // Same a the field `id` + readonly createdAt: string; // ISO string +}; + +// After server confirms the subscription +export type ConfirmedSubscription = Omit & { + readonly status: 'confirmed'; + readonly rpcSubscriptionId: number; // Server's confirmation ID + readonly confirmedAt: string; // ISO string +}; + +// Union type for all states +export type Subscription = PendingSubscription | ConfirmedSubscription; diff --git a/packages/snap/src/index.ts b/packages/snap/src/index.ts index 938d1800f..d6f39a372 100644 --- a/packages/snap/src/index.ts +++ b/packages/snap/src/index.ts @@ -1,16 +1,19 @@ import { KeyringRpcMethod } from '@metamask/keyring-api'; import { handleKeyringRequest } from '@metamask/keyring-snap-sdk'; import type { - GetClientStatusResult, Json, OnAssetHistoricalPriceHandler, OnAssetsConversionHandler, OnAssetsLookupHandler, OnClientRequestHandler, OnCronjobHandler, + OnInstallHandler, OnKeyringRequestHandler, OnProtocolRequestHandler, + OnStartHandler, + OnUpdateHandler, OnUserInputHandler, + OnWebSocketEventHandler, } from '@metamask/snaps-sdk'; import { MethodNotFoundError, @@ -37,9 +40,10 @@ import { eventHandlers as confirmSignMessageEvents } from './features/confirmati import { eventHandlers as confirmSignAndSendTransactionEvents } from './features/confirmation/views/ConfirmTransactionRequest/events'; import { eventHandlers as sendFormEvents } from './features/send/views/SendForm/events'; import { eventHandlers as transactionConfirmationEvents } from './features/send/views/TransactionConfirmation/events'; -import { installPolyfills } from './infrastructure'; +import { installPolyfills } from './polyfills'; import snapContext, { clientRequestHandler, + eventEmitter, keyring, state, } from './snapContext'; @@ -264,3 +268,17 @@ export const onClientRequest: OnClientRequestHandler = async ({ request }) => { ); return result ?? null; }; + +export const onWebSocketEvent: OnWebSocketEventHandler = async ({ event }) => + withCatchAndThrowSnapError(async () => + eventEmitter.emitSync('onWebSocketEvent', event), + ); + +export const onStart: OnStartHandler = async () => + withCatchAndThrowSnapError(async () => eventEmitter.emitSync('onStart')); + +export const onUpdate: OnUpdateHandler = async () => + withCatchAndThrowSnapError(async () => eventEmitter.emitSync('onUpdate')); + +export const onInstall: OnInstallHandler = async () => + withCatchAndThrowSnapError(async () => eventEmitter.emitSync('onInstall')); diff --git a/packages/snap/src/infrastructure/event-emitter/EventEmitter.test.ts b/packages/snap/src/infrastructure/event-emitter/EventEmitter.test.ts new file mode 100644 index 000000000..3065c69a9 --- /dev/null +++ b/packages/snap/src/infrastructure/event-emitter/EventEmitter.test.ts @@ -0,0 +1,118 @@ +import { mockLogger } from '../../core/services/mocks/logger'; +import { EventEmitter } from './EventEmitter'; + +describe('EventEmitter', () => { + let eventEmitter: EventEmitter; + + beforeEach(() => { + eventEmitter = new EventEmitter(mockLogger); + }); + + describe('on', () => { + it('adds a listener for an event', () => { + const listener = jest.fn(); + + eventEmitter.on('someEvent', listener); + + expect(eventEmitter.listenerCount('someEvent')).toBe(1); + }); + + it('adds multiple listeners for an event', () => { + const listener1 = jest.fn(); + const listener2 = jest.fn(); + + eventEmitter.on('someEvent', listener1); + eventEmitter.on('someEvent', listener2); + + expect(eventEmitter.listenerCount('someEvent')).toBe(2); + }); + }); + + describe('off', () => { + it('removes a listener for an event', () => { + const listener = jest.fn(); + eventEmitter.on('someEvent', listener); + + eventEmitter.off('someEvent', listener); + + expect(eventEmitter.listenerCount('someEvent')).toBe(0); + }); + + it('removes multiple listeners for an event', () => { + const listener1 = jest.fn(); + const listener2 = jest.fn(); + eventEmitter.on('someEvent', listener1); + eventEmitter.on('someEvent', listener2); + + eventEmitter.off('someEvent', listener1); + + expect(eventEmitter.listenerCount('someEvent')).toBe(1); + }); + }); + + describe('emit', () => { + it('calls all listeners for an event', async () => { + const listener1 = jest.fn(); + const listener2 = jest.fn(); + const listener3 = jest.fn(); + eventEmitter.on('someEvent', listener1); + eventEmitter.on('someEvent', listener2); + eventEmitter.on('someEvent', listener3); + + await eventEmitter.emitSync('someEvent'); + + expect(listener1).toHaveBeenCalled(); + expect(listener2).toHaveBeenCalled(); + expect(listener3).toHaveBeenCalled(); + }); + + it('does not call listeners for a different event', async () => { + const listener1 = jest.fn(); + const listener2 = jest.fn(); + eventEmitter.on('someEvent', listener1); + eventEmitter.on('someOtherEvent', listener2); + + await eventEmitter.emitSync('someEvent'); + + expect(listener1).toHaveBeenCalled(); + expect(listener2).not.toHaveBeenCalled(); + }); + }); + + describe('removeAllListeners', () => { + it('removes all listeners for an event', () => { + const listener1 = jest.fn(); + const listener2 = jest.fn(); + eventEmitter.on('someEvent', listener1); + eventEmitter.on('someEvent', listener2); + + eventEmitter.removeAllListeners('someEvent'); + + expect(eventEmitter.listenerCount('someEvent')).toBe(0); + }); + + it('removes all listeners for all events', () => { + const listener1 = jest.fn(); + const listener2 = jest.fn(); + eventEmitter.on('someEvent', listener1); + eventEmitter.on('someOtherEvent', listener2); + + eventEmitter.removeAllListeners(); + + expect(eventEmitter.listenerCount('someEvent')).toBe(0); + expect(eventEmitter.listenerCount('someOtherEvent')).toBe(0); + }); + }); + + describe('eventNames', () => { + it('returns the names of all events', () => { + eventEmitter.on('someEvent', jest.fn()); + eventEmitter.on('someOtherEvent', jest.fn()); + + expect(eventEmitter.eventNames()).toStrictEqual([ + 'someEvent', + 'someOtherEvent', + ]); + }); + }); +}); diff --git a/packages/snap/src/infrastructure/event-emitter/EventEmitter.ts b/packages/snap/src/infrastructure/event-emitter/EventEmitter.ts new file mode 100644 index 000000000..4e1af32e1 --- /dev/null +++ b/packages/snap/src/infrastructure/event-emitter/EventEmitter.ts @@ -0,0 +1,122 @@ +import type { ILogger } from '../../core/utils/logger'; + +type Listener = (data?: any) => Promise; + +/** + * The EventEmitter class is a simple event emitter that allows to register listeners / callbacks for events and emit events. + * + * @example + * ```ts + * const eventEmitter = new EventEmitter(); + * eventEmitter.on('someEvent', () => { + * someFunction(); + * }); + * eventEmitter.emit('someEvent'); + * // 'someFunction' will be called + * ``` + */ +export class EventEmitter { + readonly #logger: ILogger; + + readonly #loggerPrefix = '[⚡ EventEmitter]'; + + readonly #listeners: Map> = new Map(); + + constructor(logger: ILogger) { + this.#logger = logger; + } + + /** + * Registers a listener for an event. + * @param event - The event to listen to. + * @param listener - The listener to call when the event is emitted. + */ + on(event: string, listener: Listener) { + this.#logger.info(this.#loggerPrefix, `Adding listener for event ${event}`); + + if (!this.#listeners.has(event)) { + this.#listeners.set(event, new Set()); + } + + this.#listeners.get(event)?.add(listener); + } + + /** + * Removes a listener for an event. + * @param event - The event to remove the listener for. + * @param listener - The listener to remove. + */ + off(event: string, listener: Listener) { + this.#logger.info( + this.#loggerPrefix, + `Removing listener for event ${event}`, + ); + + const listeners = this.#listeners.get(event); + if (listeners) { + listeners.delete(listener); + + // Clean up empty sets + if (listeners.size === 0) { + this.#listeners.delete(event); + } + } + } + + /** + * Emits an event synchronously, and waits for all listeners to complete. + * The event is emitted to all listeners, and the function returns when all listeners have completed. + * Because of how the snaps platform works, we MUST await for all listeners to complete, otherwise the snap execution will stop. + * @param event - The event to emit. + * @param data - The data to pass to the listeners. + */ + async emitSync(event: string, data?: any): Promise { + this.#logger.info(this.#loggerPrefix, `Emitting event ${event}`); + const listeners = this.#listeners.get(event); + + if (listeners) { + const promises = Array.from(listeners).map( + async (listener) => await listener(data), + ); + await Promise.allSettled(promises); + } + } + + /** + * Removes all listeners for an event. + * @param event - The event to remove the listeners for. If not provided, all listeners will be removed. + */ + removeAllListeners(event?: string) { + if (event) { + this.#logger.info( + this.#loggerPrefix, + `Removing all listeners for event ${event}`, + ); + this.#listeners.delete(event); + } else { + this.#logger.info( + this.#loggerPrefix, + `Removing all listeners for all events`, + ); + this.#listeners.clear(); + } + } + + /** + * Returns the number of listeners for an event. + * @param event - The event to get the listener count for. + * @returns The number of listeners for the event. + */ + listenerCount(event: string): number { + const listeners = this.#listeners.get(event); + return listeners ? listeners.size : 0; + } + + /** + * Returns the names of all events that have listeners. + * @returns The names of all events that have listeners. + */ + eventNames(): string[] { + return Array.from(this.#listeners.keys()); + } +} diff --git a/packages/snap/src/infrastructure/event-emitter/index.ts b/packages/snap/src/infrastructure/event-emitter/index.ts new file mode 100644 index 000000000..1a0d45706 --- /dev/null +++ b/packages/snap/src/infrastructure/event-emitter/index.ts @@ -0,0 +1 @@ +export * from './EventEmitter'; diff --git a/packages/snap/src/infrastructure/index.ts b/packages/snap/src/infrastructure/index.ts index 68da7383a..630f30508 100644 --- a/packages/snap/src/infrastructure/index.ts +++ b/packages/snap/src/infrastructure/index.ts @@ -1 +1,2 @@ -export * from './polyfills'; +export * from './event-emitter'; +export * from './subscriptions'; diff --git a/packages/snap/src/infrastructure/subscriptions/ConnectionManagerAdapter.test.ts b/packages/snap/src/infrastructure/subscriptions/ConnectionManagerAdapter.test.ts new file mode 100644 index 000000000..331e72a87 --- /dev/null +++ b/packages/snap/src/infrastructure/subscriptions/ConnectionManagerAdapter.test.ts @@ -0,0 +1,271 @@ +import { Network } from '../../core/constants/solana'; +import type { ConfigProvider } from '../../core/services/config'; +import type { + Config, + NetworkWithRpcUrls, +} from '../../core/services/config/ConfigProvider'; +import { mockLogger } from '../../core/services/mocks/logger'; +import { EventEmitter } from '../event-emitter/EventEmitter'; +import { ConnectionManagerAdapter } from './ConnectionManagerAdapter'; +import type { ConnectionRepository } from './ConnectionRepository'; + +const mockWebSocketUrl = 'wss://some-mock-url.com/ws/v3/some-id'; +const mockConnectionId = 'mock-connection-id'; + +const createMockConnection = ( + id = mockConnectionId, + url = mockWebSocketUrl, +) => ({ + id, + url, + protocols: [], +}); + +describe('ConnectionManagerAdapter', () => { + let connectionManager: ConnectionManagerAdapter; + let mockConnectionRepository: ConnectionRepository; + let mockConfigProvider: ConfigProvider; + let mockEventEmitter: EventEmitter; + + const mockNetworksConfig = [ + { + caip2Id: Network.Mainnet, + webSocketUrl: mockWebSocketUrl, + }, + { + caip2Id: Network.Devnet, + webSocketUrl: 'wss://some-mock-url2.com/ws/v3/some-id', + }, + { + caip2Id: Network.Testnet, + webSocketUrl: 'wss://some-mock-url3.com/ws/v3/some-id', + }, + { + caip2Id: Network.Localnet, + webSocketUrl: 'wss://some-mock-url4.com/ws/v3/some-id', + }, + ] as NetworkWithRpcUrls[]; + + beforeEach(() => { + jest.clearAllMocks(); + + mockConnectionRepository = { + getAll: jest.fn(), + getById: jest.fn(), + findByUrl: jest.fn(), + save: jest.fn(), + delete: jest.fn(), + getIdByUrl: jest.fn(), + getUrlById: jest.fn(), + } as unknown as ConnectionRepository; + + mockConfigProvider = { + get: jest.fn().mockReturnValue({ + networks: mockNetworksConfig, + subscription: { + maxReconnectAttempts: 5, + reconnectDelayMilliseconds: 1, // To speed up the tests + }, + }), + getNetworkBy: jest.fn().mockImplementation((key, value) => { + return mockNetworksConfig.find( + (item) => item[key as keyof NetworkWithRpcUrls] === value, + ); + }), + } as unknown as ConfigProvider; + + mockEventEmitter = new EventEmitter(mockLogger); + + connectionManager = new ConnectionManagerAdapter( + mockConnectionRepository, + mockConfigProvider, + mockEventEmitter, + mockLogger, + ); + }); + + describe('setupAllConnections', () => { + it('opens the connections for the active networks that are not already open', async () => { + jest.spyOn(mockConfigProvider, 'get').mockReturnValue({ + activeNetworks: [Network.Mainnet, Network.Devnet], + } as unknown as Config); + + // Init with an existing connection for Mainnet. We expect to only open the connection for Devnet. + const mockConnectionMainnet = createMockConnection(); + + jest + .spyOn(mockConnectionRepository, 'getAll') + .mockResolvedValueOnce([mockConnectionMainnet]); + + jest + .spyOn(mockConnectionRepository, 'save') + .mockResolvedValueOnce(mockConnectionId); + + await connectionManager.setupAllConnections(); + + expect(mockConnectionRepository.save).toHaveBeenCalledTimes(1); + expect(mockConnectionRepository.save).toHaveBeenCalledWith( + 'wss://some-mock-url2.com/ws/v3/some-id', + ); + }); + + it('does nothing for active networks that are already open', async () => { + jest.spyOn(mockConfigProvider, 'get').mockReturnValue({ + activeNetworks: [Network.Mainnet], + } as unknown as Config); + + const mockConnection = createMockConnection(); + + jest + .spyOn(mockConnectionRepository, 'getAll') + .mockResolvedValueOnce([mockConnection]); + + await connectionManager.setupAllConnections(); + + expect(mockConnectionRepository.save).toHaveBeenCalledTimes(0); + }); + + it('closes the connections for the inactive networks that are open', async () => { + jest.spyOn(mockConfigProvider, 'get').mockReturnValue({ + activeNetworks: [], + } as unknown as Config); + + const openConnection = createMockConnection(); + + jest + .spyOn(mockConnectionRepository, 'getAll') + .mockResolvedValueOnce([openConnection]); + + jest + .spyOn(mockConnectionRepository, 'findByUrl') + .mockResolvedValueOnce(openConnection); + + await connectionManager.setupAllConnections(); + + expect(mockConnectionRepository.delete).toHaveBeenCalledTimes(1); + }); + + it('does nothing for inactive networks that are not open', async () => { + jest.spyOn(mockConfigProvider, 'get').mockReturnValue({ + activeNetworks: [], + } as unknown as Config); + + jest.spyOn(mockConnectionRepository, 'getAll').mockResolvedValueOnce([]); + + await connectionManager.setupAllConnections(); + + expect(mockConnectionRepository.delete).not.toHaveBeenCalled(); + }); + + describe('when the connection fails', () => { + beforeEach(() => { + jest.spyOn(mockConfigProvider, 'get').mockReturnValue({ + activeNetworks: [Network.Mainnet], + } as unknown as Config); + + jest + .spyOn(mockConnectionRepository, 'getAll') + .mockResolvedValueOnce([]); + }); + + it('retries until it succeeds, when attempts < 5', async () => { + jest + .spyOn(mockConnectionRepository, 'save') + .mockRejectedValueOnce(new Error('Connection failed')) // 1st call is the fail attempt + .mockResolvedValueOnce(mockConnectionId); // 2nd call is the success attempt + + await connectionManager.setupAllConnections(); + + expect(mockConnectionRepository.save).toHaveBeenCalledTimes(2); + }); + + it('retries up to the max number of attempts', async () => { + jest + .spyOn(mockConnectionRepository, 'save') + .mockRejectedValue(new Error('Connection failed')); + + await connectionManager.setupAllConnections(); + + expect(mockConnectionRepository.save).toHaveBeenCalledTimes(5); + }); + }); + }); + + describe('#handleWebSocketEvent', () => { + describe('when the event is a connect', () => { + beforeEach(async () => { + const mockConnection = createMockConnection(); + + jest + .spyOn(mockConnectionRepository, 'getById') + .mockResolvedValue(mockConnection); + + jest + .spyOn(mockConnectionRepository, 'findByUrl') + .mockResolvedValueOnce(mockConnection); + }); + + it('triggers all the recovery callbacks', async () => { + const recoveryCallback0 = jest.fn(); + const recoveryCallback1 = jest.fn(); + + connectionManager.onConnectionRecovery( + Network.Mainnet, + recoveryCallback0, + ); + connectionManager.onConnectionRecovery( + Network.Mainnet, + recoveryCallback1, + ); + + // Send the connect event + await mockEventEmitter.emitSync('onWebSocketEvent', { + id: mockConnectionId, + type: 'open', + }); + + expect(recoveryCallback0).toHaveBeenCalled(); + expect(recoveryCallback1).toHaveBeenCalled(); + }); + }); + + describe('when the event is a disconnect', () => { + beforeEach(async () => { + const mockConnection = createMockConnection(); + + jest + .spyOn(mockConnectionRepository, 'getAll') + .mockResolvedValueOnce([mockConnection]); + + jest + .spyOn(mockConnectionRepository, 'getById') + .mockResolvedValueOnce(mockConnection); + }); + + it('attempts to reconnect', async () => { + // Send the connect event + await mockEventEmitter.emitSync('onWebSocketEvent', { + id: mockConnectionId, + type: 'close', + }); + + expect(mockConnectionRepository.save).toHaveBeenCalledTimes(1); + }); + }); + }); + + describe('getConnectionIdByNetwork', () => { + it('returns the connection ID for the network', async () => { + const mockConnection = createMockConnection(); + jest + .spyOn(mockConnectionRepository, 'findByUrl') + .mockResolvedValueOnce(mockConnection); + + const connectionId = await connectionManager.getConnectionIdByNetwork( + Network.Mainnet, + ); + + expect(connectionId).toBe(mockConnectionId); + }); + }); +}); diff --git a/packages/snap/src/infrastructure/subscriptions/ConnectionManagerAdapter.ts b/packages/snap/src/infrastructure/subscriptions/ConnectionManagerAdapter.ts new file mode 100644 index 000000000..ce8617070 --- /dev/null +++ b/packages/snap/src/infrastructure/subscriptions/ConnectionManagerAdapter.ts @@ -0,0 +1,322 @@ +import type { WebSocketEvent } from '@metamask/snaps-sdk'; +import { difference } from 'lodash'; + +import { Network } from '../../core/constants/solana'; +import type { ConnectionManagerPort } from '../../core/ports'; +import type { ConfigProvider } from '../../core/services/config'; +import type { NetworkWithRpcUrls } from '../../core/services/config/ConfigProvider'; +import type { ILogger } from '../../core/utils/logger'; +import type { EventEmitter } from '../event-emitter'; +import type { ConnectionRepository } from './ConnectionRepository'; + +/** + * Manages WebSocket connections for different Solana networks, providing robust connection + * lifecycle management with automatic retry logic, reconnection handling, and connection + * state tracking. + * + * Key Features: + * - Maintains a mapping between Solana networks and their corresponding WebSocket connection IDs + * - Implements exponential backoff strategy for failed connections with configurable maximum retry attempts + * - Automatically handles disconnections and attempts reconnection with proper cleanup of stale connection mappings + * - Processes WebSocket connection events (connect, disconnect, error) and triggers appropriate recovery mechanisms + * - Converts HTTP RPC URLs to WebSocket URLs for subscription endpoints + */ +export class ConnectionManagerAdapter implements ConnectionManagerPort { + readonly #configProvider: ConfigProvider; + + readonly #connectionRepository: ConnectionRepository; + + readonly #logger: ILogger; + + readonly #loggerPrefix = '[🔌 ConnectionManagerAdapter]'; + + readonly #maxReconnectAttempts: number; + + readonly #reconnectDelayMilliseconds: number; + + readonly #connectionRecoveryCallbacks: Map Promise)[]> = + new Map(); + + constructor( + connectionRepository: ConnectionRepository, + configProvider: ConfigProvider, + eventEmitter: EventEmitter, + logger: ILogger, + ) { + const { maxReconnectAttempts, reconnectDelayMilliseconds } = + configProvider.get().subscription; + + this.#connectionRepository = connectionRepository; + this.#configProvider = configProvider; + this.#logger = logger; + this.#maxReconnectAttempts = maxReconnectAttempts; + this.#reconnectDelayMilliseconds = reconnectDelayMilliseconds; + + // When the snap starts / updates / installs, we setup all the connections + eventEmitter.on('onStart', this.setupAllConnections.bind(this)); + eventEmitter.on('onUpdate', this.setupAllConnections.bind(this)); + eventEmitter.on('onInstall', this.setupAllConnections.bind(this)); + + eventEmitter.on('onWebSocketEvent', this.#handleWebSocketEvent.bind(this)); + + // Temporary bind to enable manual testing from the test dapp + eventEmitter.on( + 'onTestSetupAllConnections', + this.setupAllConnections.bind(this), + ); + } + + async setupAllConnections(): Promise { + this.#logger.info(this.#loggerPrefix, `Setting up all connections`); + + const { activeNetworks } = this.#configProvider.get(); + const inactiveNetworks = difference(Object.values(Network), activeNetworks); + + const connections = await this.#connectionRepository.getAll(); + + const isConnectionOpen = (network: Network) => + connections.some( + (connection) => connection.url === this.#getWebSocketUrl(network), + ); + + // Open the connections for the active networks that are not already open + const openingPromises = activeNetworks + .filter((network) => !isConnectionOpen(network)) + .map(async (network) => this.#openConnection(network)); + + // Close the connections for the inactive networks that are already open + const closingPromises = inactiveNetworks + .filter(isConnectionOpen) + .map(async (network) => this.#closeConnection(network)); + + await Promise.allSettled([...openingPromises, ...closingPromises]); + } + + /** + * Opens a connection for the specified network. + * @param network - The network to open a connection for. + * @returns A promise that resolves to the connection ID. + */ + async #openConnection(network: Network): Promise { + this.#logger.info( + this.#loggerPrefix, + `Opening connection for network ${network}`, + ); + + const networkConfig = this.#configProvider.getNetworkBy('caip2Id', network); + + const { webSocketUrl } = networkConfig; + if (!webSocketUrl) { + throw new Error(`No WebSocket URL found for network ${network}`); + } + + // Check if the connection already exists + const existingConnection = + await this.#connectionRepository.findByUrl(webSocketUrl); + + if (existingConnection) { + return existingConnection.id; + } + + let attempts = 0; + + while (attempts < this.#maxReconnectAttempts) { + try { + this.#logger.info( + this.#loggerPrefix, + `Opening connection for network ${network} to ${webSocketUrl} (attempt ${attempts + 1}/${this.#maxReconnectAttempts})`, + ); + + const connectionId = + await this.#connectionRepository.save(webSocketUrl); + + return connectionId; + } catch (error) { + attempts += 1; + + if (attempts >= this.#maxReconnectAttempts) { + this.#logger.error( + this.#loggerPrefix, + `Failed to open connection after all retry attempts:`, + error, + ); + throw error; + } + + const delay = + this.#reconnectDelayMilliseconds * Math.pow(2, attempts - 1); + this.#logger.info( + this.#loggerPrefix, + `Connection attempt ${attempts} failed, retrying in ${delay}ms:`, + error, + ); + + await new Promise((resolve) => setTimeout(resolve, delay)); + } + } + + // This should never be reached, but TypeScript needs it + throw new Error('Unexpected end of openConnection method'); + } + + /** + * Closes the connection for the specified network. + * @param network - The network to close the connection for. + */ + async #closeConnection(network: Network): Promise { + this.#logger.info( + this.#loggerPrefix, + `Closing connection for network ${network}`, + ); + + const webSocketUrl = this.#getWebSocketUrl(network); + + // Early return if the connection does not exist + const existingConnection = + await this.#connectionRepository.findByUrl(webSocketUrl); + + if (!existingConnection) { + this.#logger.warn( + this.#loggerPrefix, + `Tried to close connection for network ${network} but no connection was found`, + ); + return; + } + + const connectionId = existingConnection.id; + + try { + await this.#connectionRepository.delete(connectionId); + + this.#logger.info( + this.#loggerPrefix, + `Closed connection ${connectionId}`, + ); + } catch (error) { + this.#logger.error( + this.#loggerPrefix, + `Failed to close connection:`, + error, + ); + } + } + + onConnectionRecovery(network: Network, callback: () => Promise): void { + const existingCallbacks = + this.#connectionRecoveryCallbacks.get(network) ?? []; + + this.#connectionRecoveryCallbacks.set(network, [ + ...existingCallbacks, + callback, + ]); + } + + async getConnectionIdByNetwork(network: Network): Promise { + const wsUrl = this.#getWebSocketUrl(network); + const connection = await this.#connectionRepository.findByUrl(wsUrl); + + return connection?.id ?? null; + } + + async #handleWebSocketEvent(event: WebSocketEvent): Promise { + // We only care about open and close events, that inform us about connection lifecycle + if (event.type !== 'open' && event.type !== 'close') { + return; + } + + const { id, type } = event; + const connection = await this.#connectionRepository.getById(id); + + if (!connection) { + this.#logger.warn( + this.#loggerPrefix, + `No connection found with id: ${id}`, + event, + ); + return; + } + + const network = this.#findNetworkByWebSocketUrl(connection.url)?.caip2Id; + + if (!network) { + this.#logger.warn( + this.#loggerPrefix, + `No network found matching the URL of the connection`, + connection, + ); + return; + } + + this.#logger.info( + this.#loggerPrefix, + `Handling connection event: ${type} for ${id}`, + ); + + switch (type) { + case 'open': + await this.#handleConnected(network); + break; + case 'close': + await this.#handleDisconnected(network); + break; + default: + this.#logger.warn( + this.#loggerPrefix, + `Unknown connection event type: ${type}`, + ); + } + } + + async #handleConnected(network: Network): Promise { + this.#logger.info(this.#loggerPrefix, `✅ Connected to`, network); + + this.#logger.info( + this.#loggerPrefix, + `Triggering connection recovery callbacks`, + network, + ); + + // Trigger all recovery callbacks + const recoveryPromises = + this.#connectionRecoveryCallbacks.get(network)?.map(async (callback) => { + try { + await callback(); + } catch (error) { + this.#logger.error( + this.#loggerPrefix, + `Error in connection recovery callback:`, + error, + ); + } + }) ?? []; + + await Promise.allSettled(recoveryPromises); + } + + async #handleDisconnected(network: Network): Promise { + // Attempt to reconnect + await this.#openConnection(network); + } + + /** + * Converts an HTTP RPC URL to a WebSocket URL. + * @param network - The network to get the WebSocket URL for. + * @returns The WebSocket URL. + */ + #getWebSocketUrl(network: Network): string { + const { webSocketUrl } = this.#configProvider.getNetworkBy( + 'caip2Id', + network, + ); + return webSocketUrl; + } + + /** + * Gets the network for the specified connection ID. + * @param webSocketUrl - The WebSocket URL to get the network for. + * @returns The network, or null if no network is associated with the connection ID. + */ + #findNetworkByWebSocketUrl(webSocketUrl: string): NetworkWithRpcUrls | null { + return this.#configProvider.getNetworkBy('webSocketUrl', webSocketUrl); + } +} diff --git a/packages/snap/src/infrastructure/subscriptions/ConnectionRepository.test.ts b/packages/snap/src/infrastructure/subscriptions/ConnectionRepository.test.ts new file mode 100644 index 000000000..5c437e7f9 --- /dev/null +++ b/packages/snap/src/infrastructure/subscriptions/ConnectionRepository.test.ts @@ -0,0 +1,104 @@ +import { ConnectionRepository } from './ConnectionRepository'; + +describe('ConnectionRepository', () => { + let connectionRepository: ConnectionRepository; + + beforeEach(() => { + connectionRepository = new ConnectionRepository(); + + const snap = { + request: jest.fn(), + }; + (globalThis as any).snap = snap; + }); + + describe('getAll', () => { + it('returns empty array when there are no connections', async () => { + jest.spyOn(snap, 'request').mockResolvedValue([]); + + const connections = await connectionRepository.getAll(); + + expect(connections).toStrictEqual([]); + }); + + it('returns all connections opened in the extension', async () => { + jest + .spyOn(snap, 'request') + .mockResolvedValue([{ id: '1', url: 'ws://localhost:8080' }]); + + const connections = await connectionRepository.getAll(); + expect(connections).toStrictEqual([ + { id: '1', url: 'ws://localhost:8080' }, + ]); + }); + }); + + describe('getById', () => { + it('returns null when the connection does not exist', async () => { + jest.spyOn(snap, 'request').mockResolvedValue([]); + + const connection = await connectionRepository.getById('1'); + + expect(connection).toBeNull(); + }); + + it('returns the connection when it exists', async () => { + jest + .spyOn(snap, 'request') + .mockResolvedValue([{ id: '1', url: 'ws://localhost:8080' }]); + + const connection = await connectionRepository.getById('1'); + + expect(connection).toStrictEqual({ id: '1', url: 'ws://localhost:8080' }); + }); + }); + + describe('findByUrl', () => { + it('returns null when the connection does not exist', async () => { + jest.spyOn(snap, 'request').mockResolvedValue([]); + + const connection = await connectionRepository.findByUrl( + 'ws://localhost:8080', + ); + + expect(connection).toBeNull(); + }); + + it('returns the connection when it exists', async () => { + jest + .spyOn(snap, 'request') + .mockResolvedValue([{ id: '1', url: 'ws://localhost:8080' }]); + + const connection = await connectionRepository.findByUrl( + 'ws://localhost:8080', + ); + + expect(connection).toStrictEqual({ id: '1', url: 'ws://localhost:8080' }); + }); + }); + + describe('save', () => { + it('saves the connection', async () => { + jest.spyOn(snap, 'request').mockResolvedValue('1'); + + const connectionId = await connectionRepository.save( + 'ws://localhost:8080', + ); + + expect(connectionId).toBe('1'); + }); + }); + + describe('delete', () => { + it('deletes the connection', async () => { + jest.spyOn(snap, 'request').mockResolvedValue(null); + + await connectionRepository.delete('1'); + + expect(snap.request).toHaveBeenCalledWith({ + method: 'snap_closeWebSocket', + params: { id: '1' }, + }); + }); + }); +}); diff --git a/packages/snap/src/infrastructure/subscriptions/ConnectionRepository.ts b/packages/snap/src/infrastructure/subscriptions/ConnectionRepository.ts new file mode 100644 index 000000000..4c4094a4e --- /dev/null +++ b/packages/snap/src/infrastructure/subscriptions/ConnectionRepository.ts @@ -0,0 +1,65 @@ +import type { GetWebSocketsResult } from '@metamask/snaps-sdk'; + +/** + * Repository that is treating the Snap's WebSocket storage as a persistent data store, where: + * - snap.request('snap_getWebSockets') = SELECT query + * - snap.request('snap_openWebSocket') = INSERT operation + * - snap.request('snap_closeWebSocket') = DELETE operation + * + * It also tracks bidirectional mappings between the connection ID and the URL to perform fast lookups. + */ +export class ConnectionRepository { + async getAll(): Promise { + return snap.request({ + method: 'snap_getWebSockets', + }); + } + + async getById( + connectionId: string, + ): Promise { + const existingConnections = await this.getAll(); + return ( + existingConnections.find( + (connection) => connection.id === connectionId, + ) ?? null + ); + } + + async findByUrl(url: string): Promise { + const existingConnections = await this.getAll(); + + return ( + existingConnections.find((connection) => connection.url === url) ?? null + ); + } + + /** + * Creates a new connection to the specified URL. + * @param url - The URL of the connection. + * @param protocols - The protocols of the connection. + * @returns The connection ID. + */ + async save(url: string, protocols?: string[]) { + const connectionId = await snap.request({ + method: 'snap_openWebSocket', + params: { + url, + ...(protocols ? { protocols } : {}), + }, + }); + + return connectionId; + } + + /** + * Closes the connection with the specified ID. + * @param connectionId - The ID of the connection to close. + */ + async delete(connectionId: string) { + await snap.request({ + method: 'snap_closeWebSocket', + params: { id: connectionId }, + }); + } +} diff --git a/packages/snap/src/infrastructure/subscriptions/SubscriberAdapter.test.ts b/packages/snap/src/infrastructure/subscriptions/SubscriberAdapter.test.ts new file mode 100644 index 000000000..929cccb4e --- /dev/null +++ b/packages/snap/src/infrastructure/subscriptions/SubscriberAdapter.test.ts @@ -0,0 +1,805 @@ +import type { WebSocketMessage } from '@metamask/snaps-sdk'; + +import { Network } from '../../core/constants/solana'; +import type { + ConnectionManagerPort, + SubscriptionCallbacks, +} from '../../core/ports'; +import { mockLogger } from '../../core/services/mocks/logger'; +import type { + ConfirmedSubscription, + PendingSubscription, + SubscriptionRequest, +} from '../../entities'; +import { EventEmitter } from '../event-emitter'; +import { SubscriberAdapter } from './SubscriberAdapter'; +import type { SubscriptionRepository } from './SubscriptionRepository'; + +const createMockSubscriptionRequest = ( + method = 'some-method', + unsubscribeMethod = 'some-unsubscribe-method', + params = [], + network = Network.Mainnet, +): SubscriptionRequest => ({ + method, + unsubscribeMethod, + params, + network, +}); + +const createMockSubscriptionCallbacks = ( + onNotification = jest.fn(), + onSubscriptionFailed = jest.fn(), + onConnectionRecovery = jest.fn(), +): SubscriptionCallbacks => ({ + onNotification, + onSubscriptionFailed, + onConnectionRecovery, +}); + +const createMockConfirmationMessage = ( + id: string = globalThis.crypto.randomUUID(), + rpcSubscriptionId = 98765, +): WebSocketMessage => ({ + type: 'message', + id, + origin: 'some-origin', + data: { + type: 'text', + message: JSON.stringify({ + jsonrpc: '2.0', + id, + result: rpcSubscriptionId, + }), + }, +}); + +const createMockNotification = ( + id: string = globalThis.crypto.randomUUID(), + rpcSubscriptionId = 98765, + result: any = {}, +) => ({ + type: 'message', + id, + origin: 'some-origin', + data: { + type: 'text', + message: JSON.stringify({ + jsonrpc: '2.0', + method: 'some-method', + params: { subscription: rpcSubscriptionId, result }, + }), + }, +}); + +const createMockFailure = ( + error: any, + id: string = globalThis.crypto.randomUUID(), +) => ({ + type: 'message', + id, + data: { + type: 'text', + message: JSON.stringify({ + jsonrpc: '2.0', + id, + error, + }), + }, +}); + +const simulateDisconnection = async ( + eventEmitter: EventEmitter, + connectionId: string, +) => { + await eventEmitter.emitSync('onWebSocketEvent', { + event: { + type: 'close', + id: connectionId, + }, + }); +}; + +const simulateReconnection = async ( + eventEmitter: EventEmitter, + connectionId: string, +) => { + await eventEmitter.emitSync('onWebSocketEvent', { + event: { + type: 'open', + id: connectionId, + }, + }); +}; + +const triggerConnectionRecoveryCallbacks = async ( + connectionRecoveryCallbacks: Map Promise)[]>, + network: Network, +) => { + const recoveryCallbacks = connectionRecoveryCallbacks.get(network) ?? []; + await Promise.all( + recoveryCallbacks.map(async (callback) => { + await callback(); + }), + ); +}; + +describe('SubscriberAdapter', () => { + let subscriptionManager: SubscriberAdapter; + let mockConnectionManager: ConnectionManagerPort; + let mockSubscriptionRepository: SubscriptionRepository; + let mockEventEmitter: EventEmitter; + let loggerScope: string; + + const mockNetwork = Network.Mainnet; + const mockConnectionId = 'some-connection-id'; + + beforeEach(() => { + jest.clearAllMocks(); + + const snap = { + request: jest.fn(), + }; + (globalThis as any).snap = snap; + + mockConnectionManager = { + getConnectionIdByNetwork: jest.fn().mockReturnValue(mockConnectionId), + onConnectionRecovery: jest.fn(), + handleConnectionEvent: jest.fn(), + } as unknown as ConnectionManagerPort; + + mockSubscriptionRepository = { + getAll: jest.fn(), + getById: jest.fn(), + save: jest.fn(), + update: jest.fn(), + delete: jest.fn(), + findBy: jest.fn(), + } as unknown as SubscriptionRepository; + + mockEventEmitter = new EventEmitter(mockLogger); + + subscriptionManager = new SubscriberAdapter( + mockConnectionManager, + mockSubscriptionRepository, + mockEventEmitter, + mockLogger, + ); + + loggerScope = subscriptionManager.loggerPrefix; + }); + + describe('subscribe', () => { + it('persists the subscription in state', async () => { + const request = createMockSubscriptionRequest(); + const callbacks = createMockSubscriptionCallbacks(); + + await subscriptionManager.subscribe(request, callbacks); + + expect(mockSubscriptionRepository.save).toHaveBeenCalledWith( + expect.objectContaining({ + id: expect.any(String), + }), + ); + }); + + it('registers the subscription callbacks', async () => { + const request = createMockSubscriptionRequest(); + const callbacks = createMockSubscriptionCallbacks(); + + await subscriptionManager.subscribe(request, callbacks); + + expect(mockConnectionManager.onConnectionRecovery).toHaveBeenCalledWith( + request.network, + callbacks.onConnectionRecovery, + ); + }); + + describe('when the connection is open', () => { + it('sends a subscribe message', async () => { + jest.spyOn(snap, 'request').mockResolvedValueOnce(null); + const request = createMockSubscriptionRequest(); + const callbacks = createMockSubscriptionCallbacks(); + + const subscriptionId = await subscriptionManager.subscribe( + request, + callbacks, + ); + + expect(snap.request).toHaveBeenCalledWith({ + method: 'snap_sendWebSocketMessage', + params: { + id: mockConnectionId, + message: JSON.stringify({ + jsonrpc: '2.0', + id: subscriptionId, + method: 'some-method', + params: [], + }), + }, + }); + }); + }); + + // See 'complex flows' below for when the connection is not (yet) open. + }); + + describe('unsubscribe', () => { + it('does nothing when the subscription does not exist', async () => { + await subscriptionManager.unsubscribe('some-inexistent-id'); + + // There was no subscription so there shouldn't be a call to unsubscribe. + expect(snap.request).not.toHaveBeenCalled(); + }); + + it('unsubscribes from an active subscription', async () => { + const mockSubscriptionId = 'some-subscription-id'; + const mockConfirmedSubscription: ConfirmedSubscription = { + ...createMockSubscriptionRequest(), + id: mockSubscriptionId, + status: 'confirmed', + requestId: mockSubscriptionId, + rpcSubscriptionId: 98765, + createdAt: '2024-01-01T00:00:00.000Z', + confirmedAt: '2024-01-02T00:00:00.000Z', + }; + jest + .spyOn(mockSubscriptionRepository, 'getById') + .mockResolvedValue(mockConfirmedSubscription); + + // Unsubscribe + await subscriptionManager.unsubscribe('some-subscription-id'); + + // Verify unsubscribe was called + expect(snap.request).toHaveBeenCalledWith({ + method: 'snap_sendWebSocketMessage', + params: { + id: mockConnectionId, + message: expect.stringContaining( + '"method":"some-unsubscribe-method"', + ), + }, + }); + }); + }); + + describe('#handleWebSocketEvent', () => { + describe('when the message is a notification', () => { + describe('when there is no confirmed subscription for the message', () => { + it('logs a warning and does nothing', async () => { + jest + .spyOn(mockSubscriptionRepository, 'getById') + .mockResolvedValue(undefined); + const notification = createMockNotification(); + + await mockEventEmitter.emitSync('onWebSocketEvent', notification); + + expect(mockLogger.warn).toHaveBeenCalledWith( + loggerScope, + 'Received a notification, but no matching confirmed subscription found for RPC subscription ID: 98765.', + ); + }); + }); + + describe('when there is a confirmed subscription for the message', () => { + let request: SubscriptionRequest; + let callbacks: SubscriptionCallbacks; + + beforeEach(async () => { + request = createMockSubscriptionRequest(); + callbacks = createMockSubscriptionCallbacks(); + + const subscriptionId = await subscriptionManager.subscribe( + request, + callbacks, + ); + + const confirmationMessage = createMockConfirmationMessage( + subscriptionId, + 98765, + ); + + await mockEventEmitter.emitSync('onWebSocketEvent', { + event: confirmationMessage, + }); + + const confirmedSubscription: ConfirmedSubscription = { + ...request, + id: subscriptionId, + rpcSubscriptionId: 98765, + status: 'confirmed', + requestId: subscriptionId, + createdAt: '2024-01-01T00:00:00.000Z', + confirmedAt: '2024-01-02T00:00:00.000Z', + }; + + jest + .spyOn(mockSubscriptionRepository, 'findBy') + .mockResolvedValue(confirmedSubscription); + }); + + it('handles a notification', async () => { + const notification = createMockNotification(undefined, undefined, { + context: { Slot: 348893275 }, + value: { lamports: 116044436802 }, + }); + + await mockEventEmitter.emitSync('onWebSocketEvent', notification); + + expect(callbacks.onNotification).toHaveBeenCalledWith({ + context: { Slot: 348893275 }, + value: { lamports: 116044436802 }, + }); + }); + + it('catches errors on the subscription callback', async () => { + const error = new Error('Subscription callback error'); + jest + .spyOn(callbacks, 'onNotification') + .mockImplementation() + .mockRejectedValue(error); + const notification = createMockNotification(undefined, undefined, { + context: { Slot: 348893275 }, + value: { lamports: 116044436802 }, + }); + + await mockEventEmitter.emitSync('onWebSocketEvent', notification); + + expect(mockLogger.error).toHaveBeenCalledWith( + loggerScope, + 'Error in subscription callback for 98765:', + error, + ); + }); + }); + }); + + describe('when the message is a subscription confirmation', () => { + describe('when there is no subscription for the message', () => { + it('logs a warning and does nothing', async () => { + const message = createMockConfirmationMessage('some-subscription-id'); + + await mockEventEmitter.emitSync('onWebSocketEvent', message); + + expect(mockLogger.warn).toHaveBeenCalledWith( + loggerScope, + 'Received subscription confirmation, but no matching pending subscription found for subscription ID: some-subscription-id.', + ); + }); + }); + + describe('when there is a pending subscription for the message', () => { + let request: SubscriptionRequest; + let callbacks: SubscriptionCallbacks; + let subscriptionId: string; + let pendingSubscription: PendingSubscription; + + beforeEach(async () => { + request = createMockSubscriptionRequest(); + callbacks = createMockSubscriptionCallbacks(); + subscriptionId = await subscriptionManager.subscribe( + request, + callbacks, + ); + + pendingSubscription = { + ...request, + id: subscriptionId, + status: 'pending', + requestId: subscriptionId, + createdAt: '2024-01-01T00:00:00.000Z', + }; + + jest + .spyOn(mockSubscriptionRepository, 'getById') + .mockResolvedValue(pendingSubscription); + }); + + it('confirms the subscription', async () => { + const confirmationMessage = createMockConfirmationMessage(); + + await mockEventEmitter.emitSync( + 'onWebSocketEvent', + confirmationMessage, + ); + + // Verify the confirmation was updated to 'confirmed' + const confirmedSubscription: ConfirmedSubscription = { + ...pendingSubscription, + status: 'confirmed', + rpcSubscriptionId: 98765, + confirmedAt: expect.any(String), + }; + + expect(mockSubscriptionRepository.update).toHaveBeenCalledWith( + confirmedSubscription, + ); + + // Verify that notifications are now handled + + jest + .spyOn(mockSubscriptionRepository, 'findBy') + .mockResolvedValue(confirmedSubscription); + + const notification = createMockNotification( + confirmedSubscription.id, + 98765, + { + context: { Slot: 348893275 }, + value: { lamports: 116044436802 }, + }, + ); + + await mockEventEmitter.emitSync('onWebSocketEvent', notification); + + expect(callbacks.onNotification).toHaveBeenCalledWith({ + context: { Slot: 348893275 }, + value: { lamports: 116044436802 }, + }); + }); + }); + }); + + describe('when the message is a failure', () => { + describe('when it is a response to a specific request', () => { + const message = createMockFailure({ + code: -32000, + message: 'Subscription error', + }); + + describe('when there is a subscription for the message', () => { + let request: SubscriptionRequest; + let callbacks: SubscriptionCallbacks; + let subscriptionId: string; + + beforeEach(async () => { + request = createMockSubscriptionRequest(); // request ID is 2 (request ID 1 was for opening the connection), hence why we createMockFailure with 2 as first argument + callbacks = createMockSubscriptionCallbacks(); + subscriptionId = await subscriptionManager.subscribe( + request, + callbacks, + ); + + const pendingSubscription: PendingSubscription = { + ...request, + id: subscriptionId, + status: 'pending', + requestId: subscriptionId, + createdAt: '2024-01-01T00:00:00.000Z', + }; + + jest + .spyOn(mockSubscriptionRepository, 'getById') + .mockResolvedValue(pendingSubscription); + }); + + it('logs the error', async () => { + await mockEventEmitter.emitSync('onWebSocketEvent', message); + + expect(mockLogger.error).toHaveBeenCalledWith( + loggerScope, + `Subscription establishment failed for ${subscriptionId}:`, + { + code: -32000, + message: 'Subscription error', + }, + ); + }); + + it('calls the subscription callback with the error', async () => { + await mockEventEmitter.emitSync('onWebSocketEvent', message); + + expect(callbacks.onSubscriptionFailed).toHaveBeenCalledWith({ + code: -32000, + message: 'Subscription error', + }); + }); + }); + + describe('when there is no subscription for the message', () => { + it('logs an error and does nothing', async () => { + await mockEventEmitter.emitSync('onWebSocketEvent', message); + + expect(mockLogger.error).toHaveBeenCalledWith( + loggerScope, + `Received error for request ID: ${message.id}`, + { + code: -32000, + message: 'Subscription error', + }, + ); + }); + }); + }); + + describe('when it is a connection-level error', () => { + const message = { + type: 'message', + id: 'some-id', + origin: 'some-origin', + data: { + type: 'text', + message: JSON.stringify({ + jsonrpc: '2.0', + error: { + code: -32700, + message: 'Parse error', + }, + }), + }, + }; + + it('logs an error and does nothing', async () => { + await mockEventEmitter.emitSync('onWebSocketEvent', message); + + expect(mockLogger.error).toHaveBeenCalledWith( + loggerScope, + 'Connection-level error:', + { + code: -32700, + message: 'Parse error', + }, + ); + }); + }); + }); + }); + + describe('re-subscription scenarios', () => { + let connectionRecoveryCallbacks: Map Promise)[]>; + + beforeEach(() => { + jest.clearAllMocks(); + + // Prepare the mock connection manager to handle the reconnection + connectionRecoveryCallbacks = new Map(); + + jest + .spyOn(mockConnectionManager, 'onConnectionRecovery') + .mockImplementation((network, callback) => { + connectionRecoveryCallbacks.set(network, [ + ...(connectionRecoveryCallbacks.get(network) ?? []), + callback, + ]); + }); + }); + + describe('when a subscription request is sent, but no open connection', () => { + it('saves the subscription request, and sends it once the connection is opened', async () => { + // Mock the getConnectionIdByNetwork to simulate connection state changes + jest + .spyOn(mockConnectionManager, 'getConnectionIdByNetwork') + .mockResolvedValue(null) + .mockResolvedValueOnce(null) // First call returns null (no connection) + .mockResolvedValueOnce(mockConnectionId); // Second call returns connection ID + + const request = createMockSubscriptionRequest(); + const callbacks = createMockSubscriptionCallbacks(); + const subscriptionId = await subscriptionManager.subscribe( + request, + callbacks, + ); + + const pendingSubscription: PendingSubscription = { + ...request, + id: subscriptionId, + status: 'pending', + requestId: subscriptionId, + createdAt: expect.any(String), + }; + + // Verify that the subscription request was saved, but not sent yet + expect(mockSubscriptionRepository.save).toHaveBeenCalledWith( + pendingSubscription, + ); + expect(snap.request).not.toHaveBeenCalled(); + + // Verify that the connection recovery callback was registered + expect(mockConnectionManager.onConnectionRecovery).toHaveBeenCalledWith( + mockNetwork, + expect.any(Function), + ); + expect(connectionRecoveryCallbacks.get(mockNetwork)).toHaveLength(2); // 1 for the subscription's connection recovery callback, 1 for retrying the subscription request + expect(connectionRecoveryCallbacks.get(mockNetwork)?.[0]).toBe( + callbacks.onConnectionRecovery, + ); + + // Now, let's establish the connection + jest + .spyOn(mockConnectionManager, 'getConnectionIdByNetwork') + .mockResolvedValue(mockConnectionId); + + // Send the connection event + await simulateReconnection(mockEventEmitter, mockConnectionId); + + // Manually trigger the connection recovery callbacks since we can't mock the private method #handleWebSocketEvent + await triggerConnectionRecoveryCallbacks( + connectionRecoveryCallbacks, + mockNetwork, + ); + + // Verify that the subscription request was sent + expect(snap.request).toHaveBeenCalledWith({ + method: 'snap_sendWebSocketMessage', + params: { + id: mockConnectionId, + message: JSON.stringify({ + jsonrpc: '2.0', + id: subscriptionId, + method: 'some-method', + params: [], + }), + }, + }); + + // Verify that the onConnectionRecovery callback was called + expect(callbacks.onConnectionRecovery).toHaveBeenCalledWith(); + }); + + it('registers a connection recovery callback when the subscription has one', async () => { + const request = createMockSubscriptionRequest(); + const callbacks = createMockSubscriptionCallbacks(); + + await subscriptionManager.subscribe(request, callbacks); + + expect(mockConnectionManager.onConnectionRecovery).toHaveBeenCalledWith( + mockNetwork, + callbacks.onConnectionRecovery, + ); + }); + }); + + describe('when the connection is lost BEFORE the subscription is confirmed', () => { + it('re-sends the subscription request when the connection is reestablished', async () => { + const request = createMockSubscriptionRequest(); + const callbacks = createMockSubscriptionCallbacks(); + const subscriptionId = await subscriptionManager.subscribe( + request, + callbacks, + ); + + const pendingSubscription: PendingSubscription = { + ...request, + id: subscriptionId, + status: 'pending', + requestId: subscriptionId, + createdAt: expect.any(String), + }; + + // Verify that the subscription request was saved + expect(mockSubscriptionRepository.save).toHaveBeenCalledWith( + pendingSubscription, + ); + + // Verify that the connection request was sent + expect(snap.request).toHaveBeenCalledWith({ + method: 'snap_sendWebSocketMessage', + params: { + id: mockConnectionId, + message: JSON.stringify({ + jsonrpc: '2.0', + id: subscriptionId, + method: 'some-method', + params: [], + }), + }, + }); + + // Do not confirm the subscription yet + + // Now, simulate a disconnection (this has no direct effect, it's just for clarity of the test) + await simulateDisconnection(mockEventEmitter, mockConnectionId); + + // Simulate a reconnection + await simulateReconnection(mockEventEmitter, mockConnectionId); + + // Manually trigger the connection recovery callbacks since we can't mock the private method #handleWebSocketEvent + await triggerConnectionRecoveryCallbacks( + connectionRecoveryCallbacks, + mockNetwork, + ); + + // Verify that the subscription request was sent again + expect(snap.request).toHaveBeenCalledWith({ + method: 'snap_sendWebSocketMessage', + params: { + id: mockConnectionId, + message: JSON.stringify({ + jsonrpc: '2.0', + id: subscriptionId, + method: 'some-method', + params: [], + }), + }, + }); + + // Verify that the onConnectionRecovery callback was called + expect(callbacks.onConnectionRecovery).toHaveBeenCalledWith(); + }); + }); + + describe('when the connection is lost AFTER the subscription is confirmed', () => { + it('re-sends the subscription request when the connection is reestablished', async () => { + const request = createMockSubscriptionRequest(); + const callbacks = createMockSubscriptionCallbacks(); + const subscriptionId = await subscriptionManager.subscribe( + request, + callbacks, + ); + + const pendingSubscription: PendingSubscription = { + ...request, + id: subscriptionId, + status: 'pending', + requestId: subscriptionId, + createdAt: expect.any(String), + }; + + // Verify that the subscription request was saved + expect(mockSubscriptionRepository.save).toHaveBeenCalledWith( + pendingSubscription, + ); + + // Verify that the connection request was sent + expect(snap.request).toHaveBeenCalledWith({ + method: 'snap_sendWebSocketMessage', + params: { + id: mockConnectionId, + message: JSON.stringify({ + jsonrpc: '2.0', + id: subscriptionId, + method: 'some-method', + params: [], + }), + }, + }); + + // Confirm the subscription + const confirmationMessage = createMockConfirmationMessage( + subscriptionId, + 98765, + ); + await mockEventEmitter.emitSync( + 'onWebSocketEvent', + confirmationMessage, + ); + + jest.spyOn(mockSubscriptionRepository, 'findBy').mockResolvedValue({ + ...pendingSubscription, + status: 'confirmed', + rpcSubscriptionId: 98765, + confirmedAt: expect.any(String), + }); + + // Now, simulate a disconnection (this has no direct effect, it's just for clarity of the test) + await simulateDisconnection(mockEventEmitter, mockConnectionId); + + // Simulate a reconnection + await simulateReconnection(mockEventEmitter, mockConnectionId); + + // Verify that the subscription request was sent again + expect(snap.request).toHaveBeenCalledWith({ + method: 'snap_sendWebSocketMessage', + params: { + id: mockConnectionId, + message: JSON.stringify({ + jsonrpc: '2.0', + id: subscriptionId, + method: 'some-method', + params: [], + }), + }, + }); + + // Manually trigger the connection recovery callbacks since we can't mock the private method #handleWebSocketEvent + await triggerConnectionRecoveryCallbacks( + connectionRecoveryCallbacks, + mockNetwork, + ); + + // Verify that the onConnectionRecovery callback was called + expect(callbacks.onConnectionRecovery).toHaveBeenCalledWith(); + }); + }); + }); +}); diff --git a/packages/snap/src/infrastructure/subscriptions/SubscriberAdapter.ts b/packages/snap/src/infrastructure/subscriptions/SubscriberAdapter.ts new file mode 100644 index 000000000..b1cd47887 --- /dev/null +++ b/packages/snap/src/infrastructure/subscriptions/SubscriberAdapter.ts @@ -0,0 +1,516 @@ +import type { WebSocketEvent } from '@metamask/snaps-sdk'; +import type { JsonRpcFailure } from '@metamask/utils'; +import { isJsonRpcFailure, type JsonRpcRequest } from '@metamask/utils'; + +import { Network } from '../../core/constants/solana'; +import type { + ConnectionManagerPort, + SubscriberPort, + SubscriptionCallbacks, +} from '../../core/ports'; +import type { ILogger } from '../../core/utils/logger'; +import type { PendingSubscription, SubscriptionRequest } from '../../entities'; +import type { EventEmitter } from '../event-emitter'; +import type { SubscriptionRepository } from './SubscriptionRepository'; + +/** + * A message that we receive from the RPC WebSocket server after a subscription request, + * that confirms that the subscription was successfully established. + */ +type JsonRpcWebSocketSubscriptionConfirmation = { + jsonrpc: string; + id: string | number; + result: number; +}; + +/** + * A message that we receive from the RPC WebSocket server after a subscription is confirmed. + * It contains the notification data we subscribed to. + */ +type JsonRpcWebSocketNotification = { + jsonrpc: string; + method: string; + params: { + subscription: number; + result: any; + }; +}; + +/** + * Allows subscribing / unsubscribing from real-time notifications from the Solana blockchain using the [RPC WebSocket API](https://solana.com/docs/rpc/websocket). + * + * @example + * ```ts + * const subscriber = new SubscriberAdapter(...); + * await subscriber.subscribe(request, callbacks); + * await subscriber.unsubscribe(subscriptionId); + * ``` + */ +export class SubscriberAdapter implements SubscriberPort { + readonly #connectionManager: ConnectionManagerPort; + + readonly #subscriptionRepository: SubscriptionRepository; + + readonly #logger: ILogger; + + readonly loggerPrefix = '[🔔 SubscriberAdapter]'; + + // TODO: This is problematic because the subscriptions are persisted in the state, but not the callbacks. + readonly #callbacks: Map = new Map(); // subscription ID -> callbacks + + constructor( + connectionManager: ConnectionManagerPort, + subscriptionRepository: SubscriptionRepository, + eventEmitter: EventEmitter, + logger: ILogger, + ) { + this.#connectionManager = connectionManager; + this.#subscriptionRepository = subscriptionRepository; + this.#logger = logger; + + // When the extension starts, it has lost all its websockets, so we need to clear the subscriptions. + eventEmitter.on('onStart', this.#clearSubscriptions.bind(this)); + eventEmitter.on('onWebSocketEvent', this.#handleWebSocketEvent.bind(this)); + + // Temporary bind to enable manual testing from the test dapp + eventEmitter.on( + 'onTestSubscribeToAccount', + this.#testSubscribeToAccount.bind(this), + ); + } + + /** + * Requests a new subscription. + * - If the connection is already established, the subscription request is sent immediately. + * - If the connection is not established, the subscription request is saved and will be sent later, when the connection is established. + * - If the subscription has a connection recovery callback, it is registered with the connection manager. + * + * @param request - The subscription request. + * @param callbacks - The callbacks to call when the subscription is established or fails. + * @returns The ID of the subscription. + */ + async subscribe( + request: SubscriptionRequest, + callbacks: SubscriptionCallbacks, + ): Promise { + this.#logger.info( + this.loggerPrefix, + `New subscription request`, + request, + callbacks, + ); + + const { method, params, network } = request; + const { onConnectionRecovery } = callbacks; + + const id = this.#generateId(); + + const pendingSubscription: PendingSubscription = { + ...request, + id, + status: 'pending', + requestId: id, // Use the same ID for the request and the subscription for easier lookup. + createdAt: new Date().toISOString(), + }; + + // Before sending the request, save the subscription in the repository. + // When it gets confirmed, we will update the status to 'active'. + await this.#subscriptionRepository.save(pendingSubscription); + + this.#callbacks.set(id, callbacks); + + // If the subscription has a connection recovery callback, register it with the connection manager. + if (onConnectionRecovery) { + this.#connectionManager.onConnectionRecovery( + network, + onConnectionRecovery, + ); + } + + const connectionId = + await this.#connectionManager.getConnectionIdByNetwork(network); + + const sendSubscriptionMessage = async (_connectionId: string) => { + const message: JsonRpcRequest = { + jsonrpc: '2.0', + id, + method, + params, + }; + if (_connectionId) { + await this.#sendMessage(_connectionId, message); + } + }; + + /** + * Register a callback that will send the message when the connection is reestablished. It covers both cases: + * - The connection was lost then re-established -> we need to re-subscribe. + * - The connection was not yet established, and we need to re-subscribe when it is established. + */ + this.#connectionManager.onConnectionRecovery(network, async () => { + const futureConnectionId = + await this.#connectionManager.getConnectionIdByNetwork(network); + if (futureConnectionId) { + await sendSubscriptionMessage(futureConnectionId); + } + }); + + // If the connection is open, send the message immediately. + if (connectionId) { + await sendSubscriptionMessage(connectionId); + } + + return pendingSubscription.id; + } + + async unsubscribe(subscriptionId: string): Promise { + this.#logger.info(this.loggerPrefix, `Unsubscribing`, subscriptionId); + + // Attempt to find the subscription in the repository + const subscription = + await this.#subscriptionRepository.getById(subscriptionId); + + if (!subscription) { + this.#logger.warn( + this.loggerPrefix, + `Subscription not found: ${subscriptionId}`, + ); + return; + } + + const { id, network, unsubscribeMethod } = subscription; + + // If the subscription is active, we need to unsubscribe from the RPC and remove it from the active map. + if (subscription.status === 'confirmed') { + const connectionId = + await this.#connectionManager.getConnectionIdByNetwork(network); + + if (connectionId) { + await this.#sendMessage(connectionId, { + jsonrpc: '2.0', + id: this.#generateId(), + method: unsubscribeMethod, + params: [subscription.rpcSubscriptionId], + }); + } + } + + // Whatever the status is, we delete the subscription from the repository. + await this.#subscriptionRepository.delete(id); + } + + async #handleWebSocketEvent(message: WebSocketEvent): Promise { + // We only care about actual messages, not open or close events. + if (message.type !== 'message') { + return; + } + + this.#logger.info(this.loggerPrefix, `Received message`, message); + + try { + const { data } = message; + let parsedMessage: any; + + // Handle SIP-20 message format + if (data && typeof data === 'object' && 'type' in data) { + // This is already a SIP-20 formatted message data + if (data.type === 'text') { + parsedMessage = + typeof data.message === 'string' + ? JSON.parse(data.message) + : data.message; + } else if (data.type === 'binary') { + // Convert binary message to string and parse + const binaryArray = data.message; + const messageString = String.fromCharCode(...binaryArray); + parsedMessage = JSON.parse(messageString); + } else { + this.#logger.warn(this.loggerPrefix, `Unknown message data`, data); + return; + } + } else { + // Fallback for direct message parsing + parsedMessage = typeof data === 'string' ? JSON.parse(data) : data; + } + + if (this.#isNotification(parsedMessage)) { + await this.#handleNotification(parsedMessage); + } else if (this.#isSubscriptionConfirmation(parsedMessage)) { + await this.#handleSubscriptionConfirmation(parsedMessage); + } else if (isJsonRpcFailure(parsedMessage) || 'error' in parsedMessage) { + await this.#handleFailure(parsedMessage); + } + } catch (error) { + this.#logger.error(this.loggerPrefix, `Failed to handle message`, error); + } + } + + /** + * Sends a message to the WebSocket connection. + * @param connectionId - The ID of the connection to send the message to. + * @param message - The message to send. + * @returns A promise that resolves when the message is sent. + */ + async #sendMessage( + connectionId: string, + message: JsonRpcRequest, + ): Promise { + this.#logger.info( + this.loggerPrefix, + `Sending message to connection ${connectionId}`, + message, + ); + + await snap.request({ + method: 'snap_sendWebSocketMessage', + params: { + id: connectionId, + message: JSON.stringify(message), + }, + }); + } + + /** + * Checks if the message is a notification. + * @param message - The message to check. + * @returns True if the message is a notification, false otherwise. + */ + #isNotification(message: any): message is JsonRpcWebSocketNotification { + return ( + 'method' in message && message.method !== undefined && 'params' in message + ); + } + + /** + * Checks if the message is a subscription confirmation. + * @param message - The message to check. + * @returns True if the message is a subscription confirmation, false otherwise. + */ + #isSubscriptionConfirmation( + message: any, + ): message is JsonRpcWebSocketSubscriptionConfirmation { + return ( + 'jsonrpc' in message && + message.jsonrpc === '2.0' && + 'id' in message && + 'result' in message + ); + } + + async #handleNotification( + notification: JsonRpcWebSocketNotification, + ): Promise { + const { subscription: rpcSubscriptionId, result } = notification.params; + + const subscription = await this.#subscriptionRepository.findBy( + 'rpcSubscriptionId', + rpcSubscriptionId, + ); + + if (!subscription) { + this.#logger.warn( + this.loggerPrefix, + `Received a notification, but no matching confirmed subscription found for RPC subscription ID: ${rpcSubscriptionId}.`, + ); + + return; + } + + try { + const callbacks = this.#callbacks.get(subscription.id); + if (!callbacks) { + this.#logger.warn( + this.loggerPrefix, + `Received a notification, but no matching callbacks found for subscription ID: ${subscription.id}.`, + ); + return; + } + + await callbacks.onNotification(result); + } catch (error) { + this.#logger.error( + this.loggerPrefix, + `Error in subscription callback for ${rpcSubscriptionId}:`, + error, + ); + } + } + + async #handleSubscriptionConfirmation( + message: JsonRpcWebSocketSubscriptionConfirmation, + ): Promise { + const { id: requestId, result: rpcSubscriptionId } = message; // request ID and subscription ID are the same + + const subscription = await this.#subscriptionRepository.getById( + String(requestId), + ); + + if (!subscription) { + this.#logger.warn( + this.loggerPrefix, + `Received subscription confirmation, but no matching pending subscription found for subscription ID: ${requestId}.`, + ); + return; + } + + if (subscription.status === 'confirmed') { + this.#logger.warn( + this.loggerPrefix, + `Received subscription confirmation, but the subscription is already confirmed for request ID: ${requestId}.`, + ); + return; + } + + await this.#subscriptionRepository.update({ + ...subscription, + status: 'confirmed', + rpcSubscriptionId, + confirmedAt: new Date().toISOString(), + }); + + this.#logger.info( + this.loggerPrefix, + `Subscription confirmed: request ID: ${requestId} -> RPC ID: ${rpcSubscriptionId}`, + ); + } + + /** + * Handles the various types of WebSocket error messages. + * + * 1. Subscription Establishment Errors, when the initial subscription request fails: + * + * ```json + * // We sent: {"jsonrpc":"2.0","id":123,"method":"accountSubscribe","params":[...]} + * // Server responds with error: + * { + * "jsonrpc": "2.0", + * "error": { + * "code": -32602, + * "message": "Invalid params: account not found" + * }, + * "id": 123 // Same ID as our subscription request + * } + * ``` + * This means the subscription was never established → We should clean up from #pendingSubscriptions. + * + * 2. Individual Notification Errors. After a subscription is established, individual notifications might have errors: + * + * ```json + * // Normal notification: + * { + * "jsonrpc": "2.0", + * "method": "accountNotification", + * "params": { + * "subscription": 98765, + * "result": {"error": "Temporary RPC node issue"} + * } + * } + * ``` + * The subscription is still valid → Keep it active, just log the error. + * + * 3. Connection-Level Errors (No specific cleanup) + * ```json + * { + * "jsonrpc": "2.0", + * "error": { + * "code": -32700, + * "message": "Parse error" + * } + * // No "id" field + * } + * ``` + * + * @param response - The response to handle. + */ + async #handleFailure(response: JsonRpcFailure): Promise { + if ('id' in response && response.id !== undefined) { + // This is a response to a specific subscription request, expect the error to be related to a pending subscription. + const subscription = await this.#subscriptionRepository.getById( + String(response.id), + ); + + if (subscription?.status === 'pending') { + // The subscription request failed. We can remove the pending subscription from the repository. + await this.#subscriptionRepository.delete(subscription.id); + + this.#logger.error( + this.loggerPrefix, + `Subscription establishment failed for ${subscription.id}:`, + response.error, + ); + + const callbacks = this.#callbacks.get(subscription.id); + + // Optionally call an onSubscriptionFailed callback if the subscription has one + if (callbacks?.onSubscriptionFailed) { + try { + await callbacks.onSubscriptionFailed(response.error); + } catch (callbackError) { + this.#logger.error( + 'Error in subscription error callback:', + callbackError, + ); + } + } + } else { + // Could be an error response to an unsubscribe request or other operation + this.#logger.error( + this.loggerPrefix, + `Received error for request ID: ${response.id}`, + response.error, + ); + } + } else { + // Connection-level error - doesn't affect individual subscriptions + this.#logger.error( + this.loggerPrefix, + `Connection-level error:`, + response.error, + ); + } + } + + async #clearSubscriptions(): Promise { + this.#logger.info(this.loggerPrefix, `Clearing subscriptions`); + await this.#subscriptionRepository.deleteAll(); + } + + /** + * DELETE: Temporary method to test a subscription. + */ + async #testSubscribeToAccount(): Promise { + const subscriptionRequest: SubscriptionRequest = { + method: 'accountSubscribe', + unsubscribeMethod: 'accountUnsubscribe', + network: Network.Mainnet, + params: [ + '8A4AptCThfbuknsbteHgGKXczfJpfjuVA9SLTSGaaLGC', // LUKAzPV8dDbVykTVT14pCGKzFfNcgZgRbAXB8AGdKx3 + { commitment: 'confirmed' }, + ], + }; + + this.#logger.info( + this.loggerPrefix, + `Testing subscription`, + subscriptionRequest, + ); + + const callbacks: SubscriptionCallbacks = { + onNotification: async (message: any) => { + this.#logger.info(this.loggerPrefix, `onNotification`, message); + }, + onSubscriptionFailed: async (error: any) => { + this.#logger.info(this.loggerPrefix, `onSubscriptionFailed`, error); + }, + onConnectionRecovery: async () => { + this.#logger.info(this.loggerPrefix, `onConnectionRecovery`); + }, + }; + + await this.subscribe(subscriptionRequest, callbacks); + } + + #generateId(): string { + return globalThis.crypto.randomUUID(); + } +} diff --git a/packages/snap/src/infrastructure/subscriptions/SubscriptionRepository.test.ts b/packages/snap/src/infrastructure/subscriptions/SubscriptionRepository.test.ts new file mode 100644 index 000000000..f2430cf54 --- /dev/null +++ b/packages/snap/src/infrastructure/subscriptions/SubscriptionRepository.test.ts @@ -0,0 +1,141 @@ +import { Network } from '../../core/constants/solana'; +import type { IStateManager } from '../../core/services/state/IStateManager'; +import type { UnencryptedStateValue } from '../../core/services/state/State'; +import type { Subscription } from '../../entities'; +import { SubscriptionRepository } from './SubscriptionRepository'; + +const createMockSubscription = (id: string): Subscription => ({ + id, + network: Network.Mainnet, + status: 'pending', + requestId: id, + method: 'accountSubscribe', + unsubscribeMethod: 'accountUnsubscribe', + params: [], + createdAt: new Date().toISOString(), +}); + +describe('SubscriptionRepository', () => { + let repository: SubscriptionRepository; + let mockStateManager: IStateManager; + + beforeEach(() => { + mockStateManager = { + get: jest.fn(), + getKey: jest.fn(), + setKey: jest.fn(), + update: jest.fn(), + deleteKey: jest.fn(), + }; + + repository = new SubscriptionRepository(mockStateManager); + }); + + describe('getAll', () => { + it('returns an empty array if there are no subscriptions', async () => { + jest.spyOn(mockStateManager, 'getKey').mockResolvedValue({}); + + const subscriptions = await repository.getAll(); + expect(subscriptions).toStrictEqual([]); + }); + + it('returns the subscriptions if there are any', async () => { + const subscription0 = createMockSubscription('0'); + const subscription1 = createMockSubscription('1'); + + jest.spyOn(mockStateManager, 'getKey').mockResolvedValue({ + [subscription0.id]: subscription0, + [subscription1.id]: subscription1, + }); + + const subscriptions = await repository.getAll(); + expect(subscriptions).toStrictEqual([subscription0, subscription1]); + }); + }); + + describe('save', () => { + it('saves the subscription', async () => { + const subscription = createMockSubscription('0'); + + await repository.save(subscription); + expect(mockStateManager.setKey).toHaveBeenCalledWith( + 'subscriptions.0', + subscription, + ); + }); + }); + + describe('delete', () => { + it('deletes the subscription', async () => { + const subscription = createMockSubscription('0'); + + await repository.delete(subscription.id); + expect(mockStateManager.deleteKey).toHaveBeenCalledWith( + 'subscriptions.0', + ); + }); + }); + + describe('deleteAll', () => { + it('deletes all subscriptions', async () => { + await repository.deleteAll(); + expect(mockStateManager.deleteKey).toHaveBeenCalledWith('subscriptions'); + }); + }); + + describe('update', () => { + it('updates the subscription', async () => { + const subscription = createMockSubscription('0'); + + await repository.update(subscription); + expect(mockStateManager.setKey).toHaveBeenCalledWith( + 'subscriptions.0', + subscription, + ); + }); + }); + + describe('findBy', () => { + it('finds the subscription by the requestId', async () => { + const subscription = createMockSubscription('0'); + + jest.spyOn(mockStateManager, 'getKey').mockResolvedValue({ + [subscription.id]: subscription, + }); + + const foundSubscription = await repository.findBy( + 'requestId', + subscription.requestId, + ); + + expect(foundSubscription).toStrictEqual(subscription); + }); + + it('returns undefined if the subscription is not found', async () => { + const subscription = createMockSubscription('0'); + + jest.spyOn(mockStateManager, 'getKey').mockResolvedValue({ + [subscription.id]: subscription, + }); + + // Try to find a different subscription + const foundSubscription = await repository.findBy('requestId', '1'); + + expect(foundSubscription).toBeUndefined(); + }); + + it('returns undefined if there are no subscriptions', async () => { + jest.spyOn(mockStateManager, 'getKey').mockResolvedValue({}); + + const foundSubscription = await repository.findBy('requestId', '1'); + expect(foundSubscription).toBeUndefined(); + }); + }); + + describe('clear', () => { + it('clears all subscriptions', async () => { + await repository.clear(); + expect(mockStateManager.deleteKey).toHaveBeenCalledWith('subscriptions'); + }); + }); +}); diff --git a/packages/snap/src/infrastructure/subscriptions/SubscriptionRepository.ts b/packages/snap/src/infrastructure/subscriptions/SubscriptionRepository.ts new file mode 100644 index 000000000..ef7b1c978 --- /dev/null +++ b/packages/snap/src/infrastructure/subscriptions/SubscriptionRepository.ts @@ -0,0 +1,76 @@ +import type { IStateManager } from '../../core/services/state/IStateManager'; +import type { UnencryptedStateValue } from '../../core/services/state/State'; +import type { + ConfirmedSubscription, + PendingSubscription, + Subscription, +} from '../../entities'; + +export class SubscriptionRepository { + readonly #state: IStateManager; + + readonly #stateKey = 'subscriptions'; + + constructor(state: IStateManager) { + this.#state = state; + } + + async getAll(): Promise { + const subscriptionsById = await this.#state.getKey< + Record + >(`${this.#stateKey}`); + + return Object.values(subscriptionsById ?? {}); + } + + async getById(subscriptionId: string): Promise { + const result = await this.#state.getKey< + UnencryptedStateValue['subscriptions'][string] + >(`${this.#stateKey}.${subscriptionId}`); + + return result; + } + + async save(subscription: Subscription): Promise { + await this.#state.setKey( + `${this.#stateKey}.${subscription.id}`, + subscription, + ); + } + + async delete(subscriptionId: string): Promise { + await this.#state.deleteKey(`${this.#stateKey}.${subscriptionId}`); + } + + async deleteAll(): Promise { + await this.#state.deleteKey(`${this.#stateKey}`); + } + + async update(subscription: Subscription): Promise { + await this.#state.setKey( + `${this.#stateKey}.${subscription.id}`, + subscription, + ); + } + + async findBy( + key: keyof PendingSubscription | keyof ConfirmedSubscription, + value: string | number, + ): Promise { + const subscriptions = await this.getAll(); + + if (subscriptions.length === 0) { + return undefined; + } + + return subscriptions.find( + (subscription) => + key in subscription && + subscription[key as keyof Subscription] === value, + ); + } + + async clear(): Promise { + await this.#state.deleteKey(`${this.#stateKey}`); + } +} diff --git a/packages/snap/src/infrastructure/subscriptions/index.ts b/packages/snap/src/infrastructure/subscriptions/index.ts new file mode 100644 index 000000000..a8d7390a9 --- /dev/null +++ b/packages/snap/src/infrastructure/subscriptions/index.ts @@ -0,0 +1,4 @@ +export * from './ConnectionManagerAdapter'; +export * from './ConnectionRepository'; +export * from './SubscriberAdapter'; +export * from './SubscriptionRepository'; diff --git a/packages/snap/src/permissions.ts b/packages/snap/src/permissions.ts index 49d359f88..85cef4590 100644 --- a/packages/snap/src/permissions.ts +++ b/packages/snap/src/permissions.ts @@ -2,7 +2,10 @@ import { KeyringRpcMethod } from '@metamask/keyring-api'; import { ClientRequestMethod } from './core/handlers/onClientRequest'; import { SolanaProtocolRequestMethod } from './core/handlers/onProtocolRequest/structs'; -import { RpcRequestMethod } from './core/handlers/onRpcRequest/types'; +import { + RpcRequestMethod, + TestDappRpcRequestMethod, +} from './core/handlers/onRpcRequest/types'; import { ConfigProvider } from './core/services/config/ConfigProvider'; const prodOrigins = ['https://portfolio.metamask.io']; @@ -33,6 +36,9 @@ const dappPermissions = isDev SolanaProtocolRequestMethod.GetLatestBlockhash, // Client methods ClientRequestMethod.SignAndSendTransactionWithoutConfirmation, + // Methods specific to the test dapp + TestDappRpcRequestMethod.TestSetupAllConnections, + TestDappRpcRequestMethod.TestSubscribeToAccount, ]) : new Set([]); diff --git a/packages/snap/src/infrastructure/polyfills/dom-exception/index.ts b/packages/snap/src/polyfills/dom-exception/index.ts similarity index 100% rename from packages/snap/src/infrastructure/polyfills/dom-exception/index.ts rename to packages/snap/src/polyfills/dom-exception/index.ts diff --git a/packages/snap/src/infrastructure/polyfills/ed25519/index.ts b/packages/snap/src/polyfills/ed25519/index.ts similarity index 100% rename from packages/snap/src/infrastructure/polyfills/ed25519/index.ts rename to packages/snap/src/polyfills/ed25519/index.ts diff --git a/packages/snap/src/infrastructure/polyfills/ed25519/polyfill.ts b/packages/snap/src/polyfills/ed25519/polyfill.ts similarity index 100% rename from packages/snap/src/infrastructure/polyfills/ed25519/polyfill.ts rename to packages/snap/src/polyfills/ed25519/polyfill.ts diff --git a/packages/snap/src/infrastructure/polyfills/ed25519/utils/buffer-source-as-uint8-array.ts b/packages/snap/src/polyfills/ed25519/utils/buffer-source-as-uint8-array.ts similarity index 100% rename from packages/snap/src/infrastructure/polyfills/ed25519/utils/buffer-source-as-uint8-array.ts rename to packages/snap/src/polyfills/ed25519/utils/buffer-source-as-uint8-array.ts diff --git a/packages/snap/src/infrastructure/polyfills/ed25519/utils/is-ed25519-algorithm.ts b/packages/snap/src/polyfills/ed25519/utils/is-ed25519-algorithm.ts similarity index 100% rename from packages/snap/src/infrastructure/polyfills/ed25519/utils/is-ed25519-algorithm.ts rename to packages/snap/src/polyfills/ed25519/utils/is-ed25519-algorithm.ts diff --git a/packages/snap/src/infrastructure/polyfills/ed25519/utils/uint8-array-as-buffer-source.ts b/packages/snap/src/polyfills/ed25519/utils/uint8-array-as-buffer-source.ts similarity index 100% rename from packages/snap/src/infrastructure/polyfills/ed25519/utils/uint8-array-as-buffer-source.ts rename to packages/snap/src/polyfills/ed25519/utils/uint8-array-as-buffer-source.ts diff --git a/packages/snap/src/infrastructure/polyfills/index.ts b/packages/snap/src/polyfills/index.ts similarity index 100% rename from packages/snap/src/infrastructure/polyfills/index.ts rename to packages/snap/src/polyfills/index.ts diff --git a/packages/snap/src/infrastructure/polyfills/queue-microtask/index.ts b/packages/snap/src/polyfills/queue-microtask/index.ts similarity index 100% rename from packages/snap/src/infrastructure/polyfills/queue-microtask/index.ts rename to packages/snap/src/polyfills/queue-microtask/index.ts diff --git a/packages/snap/src/snapContext.ts b/packages/snap/src/snapContext.ts index f4f2198f7..c2ed693b2 100644 --- a/packages/snap/src/snapContext.ts +++ b/packages/snap/src/snapContext.ts @@ -4,8 +4,9 @@ import { StateCache } from './core/caching/StateCache'; import { PriceApiClient } from './core/clients/price-api/PriceApiClient'; import { SecurityAlertsApiClient } from './core/clients/security-alerts-api/SecurityAlertsApiClient'; import { TokenMetadataClient } from './core/clients/token-metadata-client/TokenMetadataClient'; -import { ClientRequestHandler } from './core/handlers/onClientRequest'; +import { ClientRequestHandler } from './core/handlers'; import { SolanaKeyring } from './core/handlers/onKeyringRequest/Keyring'; +import type { ConnectionManagerPort, SubscriberPort } from './core/ports'; import type { Serializable } from './core/serialization/types'; import { AnalyticsService } from './core/services/analytics/AnalyticsService'; import { AssetsService } from './core/services/assets/AssetsService'; @@ -25,6 +26,13 @@ import { WalletService } from './core/services/wallet/WalletService'; import logger from './core/utils/logger'; import { SendSolBuilder } from './features/send/transactions/SendSolBuilder'; import { SendSplTokenBuilder } from './features/send/transactions/SendSplTokenBuilder'; +import { + ConnectionManagerAdapter, + ConnectionRepository, + EventEmitter, + SubscriberAdapter, + SubscriptionRepository, +} from './infrastructure'; /** * Initializes all the services using dependency injection. @@ -49,9 +57,16 @@ export type SnapExecutionContext = { cache: ICache; nftService: NftService; clientRequestHandler: ClientRequestHandler; + subscriptionConnectionManager: ConnectionManagerPort; + subscriber: SubscriberPort; + subscriptionRepository: SubscriptionRepository; + eventEmitter: EventEmitter; }; const configProvider = new ConfigProvider(); + +const eventEmitter = new EventEmitter(logger); + const state = new State({ encrypted: false, defaultState: DEFAULT_UNENCRYPTED_STATE, @@ -105,6 +120,24 @@ const transactionScanService = new TransactionScanService( const confirmationHandler = new ConfirmationHandler(); +const subscriptionConnectionRepository = new ConnectionRepository(); + +const subscriptionConnectionManager = new ConnectionManagerAdapter( + subscriptionConnectionRepository, + configProvider, + eventEmitter, + logger, +); + +const subscriptionRepository = new SubscriptionRepository(state); + +const subscriber = new SubscriberAdapter( + subscriptionConnectionManager, + subscriptionRepository, + eventEmitter, + logger, +); + const keyring = new SolanaKeyring({ state, transactionsService, @@ -144,6 +177,10 @@ const snapContext: SnapExecutionContext = { confirmationHandler, nftService, clientRequestHandler, + subscriptionConnectionManager, + subscriber, + subscriptionRepository, + eventEmitter, }; export { @@ -153,12 +190,16 @@ export { configProvider, confirmationHandler, connection, + eventEmitter, keyring, nftService, priceApiClient, sendSolBuilder, sendSplTokenBuilder, state, + subscriber, + subscriptionConnectionManager, + subscriptionRepository, tokenMetadataClient, tokenMetadataService, tokenPricesService,