diff --git a/eslint-suppressions.json b/eslint-suppressions.json index 53965062ca9..1798924e10b 100644 --- a/eslint-suppressions.json +++ b/eslint-suppressions.json @@ -701,11 +701,6 @@ "count": 1 } }, - "packages/core-backend/src/index.ts": { - "no-restricted-syntax": { - "count": 4 - } - }, "packages/core-backend/src/ws/AccountActivityService.ts": { "no-restricted-syntax": { "count": 1 diff --git a/packages/core-backend/CHANGELOG.md b/packages/core-backend/CHANGELOG.md index 172cb4987bd..059655e6277 100644 --- a/packages/core-backend/CHANGELOG.md +++ b/packages/core-backend/CHANGELOG.md @@ -7,6 +7,10 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0 ## [Unreleased] +### Added + +- Add `RampsActivityService` for validated, profile-scoped ramps activity notifications over `BackendWebSocketService` (`ramps-activity.v1.`) ([#10304](https://github.com/MetaMask/core/pull/10304)) + ### Changed - Bump `uuid` from `^9.0.1` to `^11.1.1` ([#10243](https://github.com/MetaMask/core/pull/10243)) diff --git a/packages/core-backend/README.md b/packages/core-backend/README.md index c42d3a7aaf9..f6d9674952c 100644 --- a/packages/core-backend/README.md +++ b/packages/core-backend/README.md @@ -38,6 +38,7 @@ Core backend services for MetaMask, serving as the data layer between Backend se - [Constructor Options](#constructor-options-1) - [Methods](#methods-1) - [Events Published](#events-published) + - [RampsActivityService](#rampsactivityservice) ## Installation @@ -655,3 +656,33 @@ interface AccountActivityServiceOptions { - `AccountActivityService:balanceUpdated` - Real-time balance changes - `AccountActivityService:transactionUpdated` - Transaction status updates - `AccountActivityService:statusChanged` - Chain/service status changes + +### RampsActivityService + +Profile-scoped service for receiving ramps activity notifications through +`BackendWebSocketService`. It derives the +`ramps-activity.v1.` channel from +`AuthenticationController:getSessionProfile`, preferring +`canonicalProfileId` (the same identity Ramps uses as MoonPay `external_id`) +and falling back to the per-SRP `profileId`. It validates every server event at +runtime, and automatically resubscribes after WebSocket reconnects, profile +changes, and wallet unlocks. + +```typescript +const rampsActivityService = new RampsActivityService({ + messenger: rampsActivityServiceMessenger, +}); + +await rampsActivityService.init(); + +messenger.subscribe('RampsActivityService:eventReceived', (event) => { + if (event.needsFetch) { + // Refresh ramps data via RampsController (GET is source of truth). + } +}); +``` + +Published events: + +- `RampsActivityService:eventReceived` - A validated profile-scoped ramps activity event +- `RampsActivityService:statusChanged` - The backend WebSocket connection status diff --git a/packages/core-backend/src/index.ts b/packages/core-backend/src/index.ts index be6b1a8c32c..fd1f1fc2c50 100644 --- a/packages/core-backend/src/index.ts +++ b/packages/core-backend/src/index.ts @@ -41,13 +41,11 @@ export type { SubscriptionOptions, AccountActivityServiceOptions, AccountActivityServiceActions, - AllowedActions as AccountActivityServiceAllowedActions, AccountActivityServiceTransactionUpdatedEvent, AccountActivityServiceBalanceUpdatedEvent, AccountActivityServiceSubscriptionErrorEvent, AccountActivityServiceStatusChangedEvent, AccountActivityServiceEvents, - AllowedEvents as AccountActivityServiceAllowedEvents, AccountActivityServiceMessenger, } from './ws/AccountActivityService.js'; @@ -64,6 +62,27 @@ export type { AccountActivityMessage, } from './types.js'; +// ============================================================================ +// RAMPS ACTIVITY SERVICE +// ============================================================================ + +export { + RampsActivityService, + RAMPS_ACTIVITY_SERVICE_ALLOWED_ACTIONS, + RAMPS_ACTIVITY_SERVICE_ALLOWED_EVENTS, +} from './ws/RampsActivityService.js'; + +export type { + RampsActivityEntity, + RampsActivityEvent, + RampsActivityServiceOptions, + RampsActivityServiceActions, + RampsActivityServiceEventReceivedEvent, + RampsActivityServiceStatusChangedEvent, + RampsActivityServiceEvents, + RampsActivityServiceMessenger, +} from './ws/RampsActivityService.js'; + // ============================================================================ // API PLATFORM CLIENT SERVICE // ============================================================================ @@ -96,12 +115,10 @@ export type { OHLCVSystemNotificationData, OHLCVServiceOptions, OHLCVServiceActions, - OHLCVServiceAllowedActions, OHLCVServiceBarUpdatedEvent, OHLCVServiceChainStatusChangedEvent, OHLCVServiceSubscriptionErrorEvent, OHLCVServiceEvents, - OHLCVServiceAllowedEvents, OHLCVServiceMessenger, } from './ws/ohlcv/index.js'; diff --git a/packages/core-backend/src/ws/RampsActivityService.test.ts b/packages/core-backend/src/ws/RampsActivityService.test.ts new file mode 100644 index 00000000000..c8bf1709879 --- /dev/null +++ b/packages/core-backend/src/ws/RampsActivityService.test.ts @@ -0,0 +1,538 @@ +import type { KeyringControllerUnlockEvent } from '@metamask/keyring-controller'; +import { Messenger, MOCK_ANY_NAMESPACE } from '@metamask/messenger'; +import type { + MessengerActions, + MessengerEvents, + MockAnyNamespace, +} from '@metamask/messenger'; + +import { flushPromises } from '../../../../tests/helpers.js'; +import type { ServerNotificationMessage } from './BackendWebSocketService.js'; +import { WebSocketState } from './BackendWebSocketService.js'; +import { + RampsActivityService, + RAMPS_ACTIVITY_SERVICE_ALLOWED_ACTIONS, + RAMPS_ACTIVITY_SERVICE_ALLOWED_EVENTS, +} from './RampsActivityService.js'; +import type { RampsActivityServiceMessenger } from './RampsActivityService.js'; + +type AllActions = MessengerActions; +type AllEvents = MessengerEvents; +type RootMessenger = Messenger< + MockAnyNamespace, + AllActions, + AllEvents | KeyringControllerUnlockEvent +>; + +const CONNECTION_INFO = { + state: WebSocketState.CONNECTED, + url: 'ws://test', + reconnectAttempts: 0, + timeout: 10_000, + reconnectDelay: 500, + maxReconnectDelay: 5_000, + requestTimeout: 30_000, +}; + +const PROFILE = { + identifierId: 'identifier-id', + profileId: 'profile-id', + canonicalProfileId: 'canonical-profile-id', + metaMetricsId: 'metrics-id', +}; + +const VALID_EVENT = { + eventId: 'event-id', + type: 'transaction.updated', + category: 'transaction', + occurredAt: '2026-09-17T12:00:00.000Z', + entity: { + id: 'transaction-id', + kind: 'deposit', + status: 'completed', + transactionStatus: 'confirmed', + transactionHash: '0x123', + }, + needsFetch: true, +} as const; + +const completeAsyncOperations = async (): Promise => { + await flushPromises(); + await flushPromises(); + await flushPromises(); +}; + +type ServiceSetup = { + service: RampsActivityService; + messenger: RampsActivityServiceMessenger; + rootMessenger: RootMessenger; + mocks: { + getSessionProfile: jest.Mock; + connect: jest.Mock; + subscribe: jest.Mock; + getConnectionInfo: jest.Mock; + channelHasSubscription: jest.Mock; + findSubscriptionsByChannelPrefix: jest.Mock; + }; +}; + +const setupService = (profile = PROFILE): ServiceSetup => { + const rootMessenger: RootMessenger = new Messenger({ + namespace: MOCK_ANY_NAMESPACE, + }); + const messenger: RampsActivityServiceMessenger = new Messenger({ + namespace: 'RampsActivityService', + parent: rootMessenger, + }); + + rootMessenger.delegate({ + messenger, + actions: [...RAMPS_ACTIVITY_SERVICE_ALLOWED_ACTIONS], + events: [...RAMPS_ACTIVITY_SERVICE_ALLOWED_EVENTS], + }); + + const getSessionProfile = jest.fn().mockResolvedValue(profile); + const connect = jest.fn().mockResolvedValue(undefined); + const subscribe = jest.fn().mockResolvedValue(undefined); + const getConnectionInfo = jest.fn().mockReturnValue(CONNECTION_INFO); + const channelHasSubscription = jest.fn().mockReturnValue(false); + const findSubscriptionsByChannelPrefix = jest.fn().mockReturnValue([]); + + rootMessenger.registerActionHandler( + 'AuthenticationController:getSessionProfile', + getSessionProfile, + ); + rootMessenger.registerActionHandler( + 'BackendWebSocketService:connect', + connect, + ); + rootMessenger.registerActionHandler( + 'BackendWebSocketService:subscribe', + subscribe, + ); + rootMessenger.registerActionHandler( + 'BackendWebSocketService:getConnectionInfo', + getConnectionInfo, + ); + rootMessenger.registerActionHandler( + 'BackendWebSocketService:channelHasSubscription', + channelHasSubscription, + ); + rootMessenger.registerActionHandler( + 'BackendWebSocketService:findSubscriptionsByChannelPrefix', + findSubscriptionsByChannelPrefix, + ); + + const service = new RampsActivityService({ messenger }); + + return { + service, + messenger, + rootMessenger, + mocks: { + getSessionProfile, + connect, + subscribe, + getConnectionInfo, + channelHasSubscription, + findSubscriptionsByChannelPrefix, + }, + }; +}; + +const getSubscriptionCallback = ( + subscribe: jest.Mock, +): ((notification: ServerNotificationMessage) => void) => + subscribe.mock.calls.at(-1)[0].callback; + +describe('RampsActivityService', () => { + it('derives the channel from the session profile', async () => { + const { service, mocks } = setupService(); + + await service.init(); + + expect(mocks.connect).toHaveBeenCalledTimes(1); + expect(mocks.getSessionProfile).toHaveBeenCalledTimes(1); + expect(mocks.subscribe).toHaveBeenCalledWith({ + channels: ['ramps-activity.v1.canonical-profile-id'], + channelType: 'ramps-activity.v1', + callback: expect.any(Function), + }); + }); + + it('does not subscribe when no profile can be resolved', async () => { + const { service, mocks } = setupService(); + mocks.getSessionProfile.mockRejectedValue(new Error('No profile')); + + await service.init(); + + expect(mocks.subscribe).not.toHaveBeenCalled(); + }); + + it('publishes validated events without unsupported fields', async () => { + const { service, messenger, mocks } = setupService(); + const listener = jest.fn(); + messenger.subscribe('RampsActivityService:eventReceived', listener); + await service.init(); + + getSubscriptionCallback(mocks.subscribe)({ + event: 'notification', + channel: 'ramps-activity.v1.canonical-profile-id', + data: { ...VALID_EVENT, ignored: 'field' }, + timestamp: Date.now(), + } as ServerNotificationMessage); + + expect(listener).toHaveBeenCalledWith(VALID_EVENT); + }); + + it.each([ + { ...VALID_EVENT, eventId: 1 }, + { ...VALID_EVENT, category: 1 }, + { ...VALID_EVENT, category: '' }, + { ...VALID_EVENT, entity: [] }, + { ...VALID_EVENT, entity: { status: 'missing-id' } }, + { ...VALID_EVENT, entity: { id: 'id', kind: 1 } }, + { ...VALID_EVENT, entity: { id: 'id', status: 1 } }, + { ...VALID_EVENT, entity: { id: 'id', transactionStatus: 1 } }, + { ...VALID_EVENT, entity: { id: 'id', transactionHash: 1 } }, + { ...VALID_EVENT, needsFetch: 'yes' }, + { ...VALID_EVENT, payload: {} }, + { ...VALID_EVENT, customerId: 'customer-id' }, + { ...VALID_EVENT, userId: 'user-id' }, + ])('ignores malformed event data %#', async (data) => { + const { service, messenger, mocks } = setupService(); + const listener = jest.fn(); + messenger.subscribe('RampsActivityService:eventReceived', listener); + await service.init(); + + getSubscriptionCallback(mocks.subscribe)({ + event: 'notification', + channel: 'ramps-activity.v1.canonical-profile-id', + data, + timestamp: Date.now(), + } as unknown as ServerNotificationMessage); + + expect(listener).not.toHaveBeenCalled(); + }); + + it('accepts future ramps categories without a Core release', async () => { + const { service, messenger, mocks } = setupService(); + const listener = jest.fn(); + messenger.subscribe('RampsActivityService:eventReceived', listener); + await service.init(); + const event = { ...VALID_EVENT, category: 'cex_deposit' }; + + getSubscriptionCallback(mocks.subscribe)({ + event: 'notification', + channel: 'ramps-activity.v1.canonical-profile-id', + data: event, + timestamp: Date.now(), + } as ServerNotificationMessage); + + expect(listener).toHaveBeenCalledWith(event); + }); + + it('accepts a null entity', async () => { + const { service, messenger, mocks } = setupService(); + const listener = jest.fn(); + messenger.subscribe('RampsActivityService:eventReceived', listener); + await service.init(); + + getSubscriptionCallback(mocks.subscribe)({ + event: 'notification', + channel: 'ramps-activity.v1.canonical-profile-id', + data: { ...VALID_EVENT, entity: null }, + timestamp: Date.now(), + } as ServerNotificationMessage); + + expect(listener).toHaveBeenCalledWith({ ...VALID_EVENT, entity: null }); + }); + + it('accepts an entity with only its required id', async () => { + const { service, messenger, mocks } = setupService(); + const listener = jest.fn(); + messenger.subscribe('RampsActivityService:eventReceived', listener); + await service.init(); + const event = { ...VALID_EVENT, entity: { id: 'entity-id' } }; + + getSubscriptionCallback(mocks.subscribe)({ + event: 'notification', + channel: 'ramps-activity.v1.canonical-profile-id', + data: event, + timestamp: Date.now(), + } as ServerNotificationMessage); + + expect(listener).toHaveBeenCalledWith(event); + }); + + it('resubscribes after reconnect', async () => { + const { service, rootMessenger, mocks } = setupService(); + await service.init(); + mocks.subscribe.mockClear(); + + rootMessenger.publish('BackendWebSocketService:connectionStateChanged', { + ...CONNECTION_INFO, + reconnectAttempts: 1, + }); + await completeAsyncOperations(); + + expect(mocks.subscribe).toHaveBeenCalledWith( + expect.objectContaining({ + channels: ['ramps-activity.v1.canonical-profile-id'], + }), + ); + }); + + it('unsubscribes the old profile before subscribing to a changed profile', async () => { + const { service, rootMessenger, mocks } = setupService(); + const unsubscribe = jest.fn().mockResolvedValue(undefined); + await service.init(); + mocks.findSubscriptionsByChannelPrefix.mockReturnValue([{ unsubscribe }]); + mocks.getSessionProfile.mockResolvedValue({ + ...PROFILE, + canonicalProfileId: 'new-canonical-profile-id', + }); + + rootMessenger.publish('AuthenticationController:profileSignIn', { + profileId: 'new-canonical-profile-id', + profileAliases: [], + profileIdChanged: true, + }); + await completeAsyncOperations(); + + expect(unsubscribe).toHaveBeenCalledTimes(1); + expect(mocks.subscribe).toHaveBeenLastCalledWith( + expect.objectContaining({ + channels: ['ramps-activity.v1.new-canonical-profile-id'], + }), + ); + }); + + it('resubscribes after wallet unlock', async () => { + const { rootMessenger, mocks } = setupService(); + + rootMessenger.publish('KeyringController:unlock'); + await completeAsyncOperations(); + + expect(mocks.subscribe).toHaveBeenCalledWith( + expect.objectContaining({ + channels: ['ramps-activity.v1.canonical-profile-id'], + }), + ); + }); + + it('cleans up active subscriptions on destroy', async () => { + const { service, mocks } = setupService(); + const unsubscribe = jest.fn().mockResolvedValue(undefined); + mocks.findSubscriptionsByChannelPrefix.mockReturnValue([{ unsubscribe }]); + + await service.destroy(); + + expect(mocks.findSubscriptionsByChannelPrefix).toHaveBeenCalledWith( + 'ramps-activity.v1', + ); + expect(unsubscribe).toHaveBeenCalledTimes(1); + }); + + it('publishes connection status changes', async () => { + const { messenger, rootMessenger } = setupService(); + const listener = jest.fn(); + messenger.subscribe('RampsActivityService:statusChanged', listener); + + rootMessenger.publish('BackendWebSocketService:connectionStateChanged', { + ...CONNECTION_INFO, + state: WebSocketState.DISCONNECTED, + }); + await completeAsyncOperations(); + + expect(listener).toHaveBeenCalledWith({ + status: WebSocketState.DISCONNECTED, + }); + }); + + it('cleans up subscriptions after sign out', async () => { + const { rootMessenger, mocks } = setupService(); + const unsubscribe = jest.fn().mockResolvedValue(undefined); + mocks.findSubscriptionsByChannelPrefix.mockReturnValue([{ unsubscribe }]); + + rootMessenger.publish( + 'AuthenticationController:stateChange', + { isSignedIn: false }, + [], + ); + await completeAsyncOperations(); + + expect(unsubscribe).toHaveBeenCalledTimes(1); + expect(mocks.subscribe).not.toHaveBeenCalled(); + }); + + it('falls back to the per-SRP profile id when canonical is empty', async () => { + const { service, mocks } = setupService({ + ...PROFILE, + canonicalProfileId: '', + }); + + await service.init(); + + expect(mocks.subscribe).toHaveBeenCalledWith( + expect.objectContaining({ + channels: ['ramps-activity.v1.profile-id'], + }), + ); + }); + + it('subscribes when the wallet signs in', async () => { + const { rootMessenger, mocks } = setupService(); + + rootMessenger.publish( + 'AuthenticationController:stateChange', + { isSignedIn: true }, + [], + ); + await completeAsyncOperations(); + + expect(mocks.subscribe).toHaveBeenCalledTimes(1); + }); + + it('does not resubscribe when sign-in state is unchanged', async () => { + const { rootMessenger, mocks } = setupService(); + rootMessenger.publish( + 'AuthenticationController:stateChange', + { isSignedIn: true }, + [], + ); + await completeAsyncOperations(); + mocks.subscribe.mockClear(); + + rootMessenger.publish( + 'AuthenticationController:stateChange', + { isSignedIn: true, sessionData: { token: 'rotated' } }, + [], + ); + await completeAsyncOperations(); + + expect(mocks.subscribe).not.toHaveBeenCalled(); + }); + + it('skips subscribe when the channel is already registered', async () => { + const { service, mocks } = setupService(); + mocks.channelHasSubscription.mockReturnValue(true); + + await service.init(); + + expect(mocks.subscribe).not.toHaveBeenCalled(); + }); + + it('skips subscribe while the WebSocket is disconnected', async () => { + const { service, mocks } = setupService(); + mocks.getConnectionInfo.mockReturnValue({ + ...CONNECTION_INFO, + state: WebSocketState.DISCONNECTED, + }); + + await service.init(); + + expect(mocks.subscribe).not.toHaveBeenCalled(); + }); + + it('skips subscribe when neither profile id is available', async () => { + const { service, mocks } = setupService({ + ...PROFILE, + canonicalProfileId: '', + profileId: '', + }); + + await service.init(); + + expect(mocks.subscribe).not.toHaveBeenCalled(); + }); + + it('does not subscribe when destroyed while resolving the profile', async () => { + const { service, mocks } = setupService(); + let resolveProfile: ((profile: typeof PROFILE) => void) | undefined; + mocks.getSessionProfile.mockReturnValue( + new Promise((resolve) => { + resolveProfile = resolve; + }), + ); + + const initPromise = service.init(); + await completeAsyncOperations(); + await service.destroy(); + resolveProfile?.(PROFILE); + await initPromise; + + expect(mocks.subscribe).not.toHaveBeenCalled(); + }); + + it('does not subscribe when destroyed while replacing a subscription', async () => { + const { service, mocks } = setupService(); + let resolveFirstUnsubscribe: (() => void) | undefined; + const unsubscribe = jest + .fn() + .mockImplementationOnce( + async () => + await new Promise((resolve) => { + resolveFirstUnsubscribe = resolve; + }), + ) + .mockResolvedValue(undefined); + mocks.findSubscriptionsByChannelPrefix.mockReturnValue([{ unsubscribe }]); + + const initPromise = service.init(); + await completeAsyncOperations(); + const destroyPromise = service.destroy(); + resolveFirstUnsubscribe?.(); + await Promise.all([initPromise, destroyPromise]); + + expect(mocks.connect).not.toHaveBeenCalled(); + expect(mocks.subscribe).not.toHaveBeenCalled(); + }); + + it('does not subscribe when destroyed after checking the channel', async () => { + const { service, mocks } = setupService(); + mocks.channelHasSubscription.mockImplementation(() => { + service.destroy().catch(() => undefined); + return false; + }); + + await service.init(); + + expect(mocks.subscribe).not.toHaveBeenCalled(); + }); + + it('does not replace a subscription after destroy', async () => { + const { service, rootMessenger, mocks } = setupService(); + await service.destroy(); + + rootMessenger.publish('AuthenticationController:profileSignIn', { + profileId: 'new-profile-id', + profileAliases: [], + profileIdChanged: true, + }); + await completeAsyncOperations(); + + expect(mocks.connect).not.toHaveBeenCalled(); + }); + + it('does not subscribe after destroy on reconnect', async () => { + const { service, rootMessenger, mocks } = setupService(); + await service.destroy(); + mocks.subscribe.mockClear(); + + rootMessenger.publish('BackendWebSocketService:connectionStateChanged', { + ...CONNECTION_INFO, + }); + await completeAsyncOperations(); + + expect(mocks.subscribe).not.toHaveBeenCalled(); + }); + + it('does not throw from init when connect fails', async () => { + const { service, mocks } = setupService(); + mocks.connect.mockRejectedValue(new Error('connect failed')); + + expect(await service.init()).toBeUndefined(); + expect(mocks.subscribe).not.toHaveBeenCalled(); + }); +}); diff --git a/packages/core-backend/src/ws/RampsActivityService.ts b/packages/core-backend/src/ws/RampsActivityService.ts new file mode 100644 index 00000000000..64f6da1efb2 --- /dev/null +++ b/packages/core-backend/src/ws/RampsActivityService.ts @@ -0,0 +1,350 @@ +import type { KeyringControllerUnlockEvent } from '@metamask/keyring-controller'; +import type { Messenger } from '@metamask/messenger'; +import type { AuthenticationController } from '@metamask/profile-sync-controller'; + +import { projectLogger, createModuleLogger } from '../logger.js'; +import type { BackendWebSocketServiceMethodActions } from './BackendWebSocketService-method-action-types.js'; +import type { + BackendWebSocketServiceConnectionStateChangedEvent, + ServerNotificationMessage, + WebSocketConnectionInfo, +} from './BackendWebSocketService.js'; +import { WebSocketState } from './BackendWebSocketService.js'; + +const SERVICE_NAME = 'RampsActivityService'; +const SUBSCRIPTION_NAMESPACE = 'ramps-activity.v1'; +const MESSENGER_EXPOSED_METHODS = [] as const; + +const log = createModuleLogger(projectLogger, SERVICE_NAME); + +export type RampsActivityEntity = { + id: string; + kind?: string; + status?: string; + transactionStatus?: string; + transactionHash?: string; +}; + +export type RampsActivityEvent = { + eventId: string; + type: string; + category: string; + occurredAt: string; + entity: RampsActivityEntity | null; + needsFetch: boolean; +}; + +export type RampsActivityServiceOptions = { + messenger: RampsActivityServiceMessenger; +}; + +export type RampsActivityServiceActions = never; + +export const RAMPS_ACTIVITY_SERVICE_ALLOWED_ACTIONS = [ + 'AuthenticationController:getSessionProfile', + 'BackendWebSocketService:connect', + 'BackendWebSocketService:subscribe', + 'BackendWebSocketService:getConnectionInfo', + 'BackendWebSocketService:channelHasSubscription', + 'BackendWebSocketService:findSubscriptionsByChannelPrefix', +] as const; + +export const RAMPS_ACTIVITY_SERVICE_ALLOWED_EVENTS = [ + 'AuthenticationController:stateChange', + 'AuthenticationController:profileSignIn', + 'BackendWebSocketService:connectionStateChanged', + 'KeyringController:unlock', +] as const; + +export type RampsActivityServiceAllowedActions = + | AuthenticationController.AuthenticationControllerGetSessionProfileAction + | BackendWebSocketServiceMethodActions; + +export type RampsActivityServiceEventReceivedEvent = { + type: 'RampsActivityService:eventReceived'; + payload: [RampsActivityEvent]; +}; + +export type RampsActivityServiceStatusChangedEvent = { + type: 'RampsActivityService:statusChanged'; + payload: [{ status: WebSocketState }]; +}; + +export type RampsActivityServiceEvents = + | RampsActivityServiceEventReceivedEvent + | RampsActivityServiceStatusChangedEvent; + +export type RampsActivityServiceAllowedEvents = + | AuthenticationController.AuthenticationControllerStateChangeEvent + | AuthenticationController.AuthenticationControllerProfileSignInEvent + | BackendWebSocketServiceConnectionStateChangedEvent + | KeyringControllerUnlockEvent; + +export type RampsActivityServiceMessenger = Messenger< + typeof SERVICE_NAME, + RampsActivityServiceActions | RampsActivityServiceAllowedActions, + RampsActivityServiceEvents | RampsActivityServiceAllowedEvents +>; + +const isRecord = (value: unknown): value is Record => + typeof value === 'object' && value !== null && !Array.isArray(value); + +const getHasOwnProperty = ( + value: Record, + key: string, +): boolean => Object.prototype.hasOwnProperty.call(value, key); + +const getOptionalString = ( + value: Record, + key: string, +): string | undefined | false => { + const property = value[key]; + if (property === undefined) { + return undefined; + } + return typeof property === 'string' ? property : false; +}; + +const parseEntity = (value: unknown): RampsActivityEntity | null | false => { + if (value === null) { + return null; + } + if (!isRecord(value) || typeof value.id !== 'string') { + return false; + } + + const kind = getOptionalString(value, 'kind'); + const status = getOptionalString(value, 'status'); + const transactionStatus = getOptionalString(value, 'transactionStatus'); + const transactionHash = getOptionalString(value, 'transactionHash'); + if ( + kind === false || + status === false || + transactionStatus === false || + transactionHash === false + ) { + return false; + } + + return { + id: value.id, + ...(kind === undefined ? {} : { kind }), + ...(status === undefined ? {} : { status }), + ...(transactionStatus === undefined ? {} : { transactionStatus }), + ...(transactionHash === undefined ? {} : { transactionHash }), + }; +}; + +const parseRampsActivityEvent = ( + value: unknown, +): RampsActivityEvent | undefined => { + if ( + !isRecord(value) || + typeof value.eventId !== 'string' || + typeof value.type !== 'string' || + typeof value.category !== 'string' || + value.category.length === 0 || + typeof value.occurredAt !== 'string' || + typeof value.needsFetch !== 'boolean' || + getHasOwnProperty(value, 'payload') || + getHasOwnProperty(value, 'customerId') || + getHasOwnProperty(value, 'userId') + ) { + return undefined; + } + + const entity = parseEntity(value.entity); + if (entity === false) { + return undefined; + } + + return { + eventId: value.eventId, + type: value.type, + category: value.category, + occurredAt: value.occurredAt, + entity, + needsFetch: value.needsFetch, + }; +}; + +/** + * Subscribes to profile-scoped ramps activity through + * {@link BackendWebSocketService}. The service owns no domain state; it only + * tracks its subscription lifecycle and publishes validated notifications. + */ +export class RampsActivityService { + readonly name = SERVICE_NAME; + + readonly #messenger: RampsActivityServiceMessenger; + + #isDestroyed = false; + + constructor({ messenger }: RampsActivityServiceOptions) { + this.#messenger = messenger; + this.#messenger.registerMethodActionHandlers( + this, + MESSENGER_EXPOSED_METHODS, + ); + + this.#messenger.subscribe( + 'BackendWebSocketService:connectionStateChanged', + // eslint-disable-next-line @typescript-eslint/no-misused-promises + async (connectionInfo: WebSocketConnectionInfo) => + await this.#handleConnectionStateChange(connectionInfo), + ); + this.#messenger.subscribe( + 'AuthenticationController:stateChange', + // eslint-disable-next-line @typescript-eslint/no-misused-promises + async (isSignedIn: boolean) => + await this.#handleAuthenticationStateChange(isSignedIn), + // Messenger compares selector results with `!==`. Return the boolean + // primitive so token/session writes do not resubscribe. + (state: AuthenticationController.AuthenticationControllerState) => + state.isSignedIn, + ); + this.#messenger.subscribe( + 'AuthenticationController:profileSignIn', + // eslint-disable-next-line @typescript-eslint/no-misused-promises + async () => await this.#replaceSubscription(), + ); + this.#messenger.subscribe( + 'KeyringController:unlock', + // eslint-disable-next-line @typescript-eslint/no-misused-promises + async () => await this.#replaceSubscription(), + ); + } + + /** + * Connect and subscribe for the current authentication profile. + */ + async init(): Promise { + await this.#replaceSubscription(); + } + + async #handleConnectionStateChange( + connectionInfo: WebSocketConnectionInfo, + ): Promise { + if (this.#isDestroyed) { + return; + } + + this.#messenger.publish('RampsActivityService:statusChanged', { + status: connectionInfo.state, + }); + + if (connectionInfo.state === WebSocketState.CONNECTED) { + await this.#subscribeToCurrentProfile(); + } + } + + async #handleAuthenticationStateChange(isSignedIn: boolean): Promise { + if (isSignedIn) { + await this.#replaceSubscription(); + } else { + await this.#unsubscribeAll(); + } + } + + async #replaceSubscription(): Promise { + if (this.#isDestroyed) { + return; + } + + await this.#unsubscribeAll(); + await this.#subscribeToCurrentProfile(); + } + + async #subscribeToCurrentProfile(): Promise { + if (this.#isDestroyed) { + return; + } + + try { + await this.#messenger.call('BackendWebSocketService:connect'); + const profile = await this.#messenger.call( + 'AuthenticationController:getSessionProfile', + ); + const profileId = this.#getChannelProfileId(profile); + if (!profileId || this.#isDestroyed) { + return; + } + + const channel = `${SUBSCRIPTION_NAMESPACE}.${profileId}`; + const connectionInfo = this.#messenger.call( + 'BackendWebSocketService:getConnectionInfo', + ); + if (connectionInfo.state !== WebSocketState.CONNECTED) { + return; + } + + if ( + this.#messenger.call( + 'BackendWebSocketService:channelHasSubscription', + channel, + ) + ) { + return; + } + + if (this.#isDestroyed) { + return; + } + + await this.#messenger.call('BackendWebSocketService:subscribe', { + channels: [channel], + channelType: SUBSCRIPTION_NAMESPACE, + callback: (notification: ServerNotificationMessage) => + this.#handleNotification(notification), + }); + } catch (error) { + log('Unable to subscribe to ramps activity', { error }); + } + } + + #getChannelProfileId(profile: { + canonicalProfileId?: string; + profileId?: string; + }): string | undefined { + const canonical = profile.canonicalProfileId; + if (typeof canonical === 'string' && canonical.length > 0) { + return canonical; + } + + const { profileId } = profile; + return typeof profileId === 'string' && profileId.length > 0 + ? profileId + : undefined; + } + + #handleNotification(notification: ServerNotificationMessage): void { + const event = parseRampsActivityEvent(notification.data); + if (!event) { + log('Ignoring malformed ramps activity event', { + channel: notification.channel, + }); + return; + } + + this.#messenger.publish('RampsActivityService:eventReceived', event); + } + + async #unsubscribeAll(): Promise { + const subscriptions = this.#messenger.call( + 'BackendWebSocketService:findSubscriptionsByChannelPrefix', + SUBSCRIPTION_NAMESPACE, + ); + + for (const subscription of subscriptions) { + await subscription.unsubscribe(); + } + } + + /** + * Stop future subscriptions and remove all active ramps subscriptions. + */ + async destroy(): Promise { + this.#isDestroyed = true; + await this.#unsubscribeAll(); + } +}