diff --git a/README.md b/README.md index c25fade294b..ab54de9e2f4 100644 --- a/README.md +++ b/README.md @@ -698,6 +698,7 @@ linkStyle default opacity:0.5 social_controllers --> base_controller; social_controllers --> base_data_service; social_controllers --> controller_utils; + social_controllers --> core_backend; social_controllers --> messenger; social_controllers --> profile_sync_controller; solana_test_validator_up --> local_node_utils; diff --git a/packages/social-controllers/CHANGELOG.md b/packages/social-controllers/CHANGELOG.md index 6d7226b3f33..f764e78ac77 100644 --- a/packages/social-controllers/CHANGELOG.md +++ b/packages/social-controllers/CHANGELOG.md @@ -7,6 +7,10 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0 ## [Unreleased] +### Added + +- Add `SocialRealtimeService` for subscribing to Social realtime feed events through the shared backend WebSocket connection ([#10561](https://github.com/MetaMask/core/pull/10561)) + ### Changed - Bump `@metamask/base-data-service` from `^2.0.0` to `^2.1.0` ([#10502](https://github.com/MetaMask/core/pull/10502)) diff --git a/packages/social-controllers/package.json b/packages/social-controllers/package.json index e07ff9532d5..6d8da6ce15b 100644 --- a/packages/social-controllers/package.json +++ b/packages/social-controllers/package.json @@ -53,6 +53,7 @@ "@metamask/base-controller": "^10.0.0", "@metamask/base-data-service": "^2.1.0", "@metamask/controller-utils": "^13.0.0", + "@metamask/core-backend": "^11.0.0", "@metamask/messenger": "^3.0.0", "@metamask/profile-sync-controller": "^33.0.0", "@metamask/superstruct": "^3.4.1" diff --git a/packages/social-controllers/src/SocialRealtimeService.test.ts b/packages/social-controllers/src/SocialRealtimeService.test.ts new file mode 100644 index 00000000000..47c964e79c0 --- /dev/null +++ b/packages/social-controllers/src/SocialRealtimeService.test.ts @@ -0,0 +1,469 @@ +import { WebSocketState } from '@metamask/core-backend'; +import type { + BackendWebSocketServiceMessenger, + WebSocketSubscription, +} from '@metamask/core-backend'; + +import { + SOCIAL_FEED_CHANNEL, + SOCIAL_FEED_CHANNEL_TYPE, + SocialRealtimeService, +} from './SocialRealtimeService.js'; + +describe('SocialRealtimeService', () => { + type ConnectionStateHandler = (connection: { state: WebSocketState }) => void; + + type BackendCall = jest.Mock; + + type TestHarness = { + call: BackendCall; + connectionStateHandler: () => ConnectionStateHandler | undefined; + messenger: BackendWebSocketServiceMessenger; + service: SocialRealtimeService; + unsubscribe: jest.Mock; + }; + + const event = { + version: 1 as const, + kind: 'feed-item' as const, + eventId: 'event-1', + feedItemId: 'position-1:trade-1', + revision: 1, + occurredAt: '2026-01-01T00:00:00.000Z', + data: { + positionId: 'position-1', + timestamp: 1, + }, + }; + + const createService = (enabled = true): TestHarness => { + let connectionStateHandler: ConnectionStateHandler | undefined; + const unsubscribe = jest.fn().mockResolvedValue(undefined); + const call: BackendCall = jest.fn( + (action: string, _options?: unknown): unknown => { + if (action === 'BackendWebSocketService:connect') { + return Promise.resolve(); + } + + if (action === 'BackendWebSocketService:channelHasSubscription') { + return false; + } + + if (action === 'BackendWebSocketService:getSubscriptionsByChannel') { + return []; + } + + return Promise.resolve({ unsubscribe }); + }, + ); + const messenger = { + call, + subscribe: jest.fn( + (_event: string, handler: ConnectionStateHandler): void => { + connectionStateHandler = handler; + }, + ), + }; + const service = new SocialRealtimeService({ + messenger: messenger as unknown as BackendWebSocketServiceMessenger, + isEnabled: (): boolean => enabled, + }); + + return { + call, + connectionStateHandler: (): ConnectionStateHandler | undefined => + connectionStateHandler, + messenger: messenger as unknown as BackendWebSocketServiceMessenger, + service, + unsubscribe, + }; + }; + + const getFeedCallback = ( + call: BackendCall, + ): ((notification: { channel: string; data: unknown }) => void) => { + const subscribeCall = call.mock.calls.find( + ([action]) => action === 'BackendWebSocketService:subscribe', + ); + if (!subscribeCall) { + throw new Error('The feed subscription was not created'); + } + + const [, subscribeOptions] = subscribeCall; + return ( + subscribeOptions as { + callback: (notification: { channel: string; data: unknown }) => void; + } + ).callback; + }; + + it('subscribes and forwards valid feed events', async () => { + const { call, service } = createService(); + const listener = jest.fn(); + service.addListener(listener); + + await service.setActive(true); + const handleNotification = getFeedCallback(call); + handleNotification({ + channel: SOCIAL_FEED_CHANNEL, + data: event, + }); + + expect(call).toHaveBeenCalledWith('BackendWebSocketService:connect'); + expect(call).toHaveBeenCalledWith( + 'BackendWebSocketService:subscribe', + expect.objectContaining({ + channels: [SOCIAL_FEED_CHANNEL], + channelType: SOCIAL_FEED_CHANNEL_TYPE, + }), + ); + expect(listener).toHaveBeenCalledWith(event); + }); + + it.each([ + ['version', { version: 2 }], + ['kind', { kind: 'other' }], + ['eventId', { eventId: 1 }], + ['feedItemId', { feedItemId: 1 }], + ['revision', { revision: '1' }], + ['occurredAt', { occurredAt: 1 }], + ['data', { data: null }], + ])( + 'ignores feed events with an invalid %s field', + async (_field, invalid) => { + const { call, service } = createService(); + const listener = jest.fn(); + service.addListener(listener); + + await service.setActive(true); + getFeedCallback(call)({ + channel: SOCIAL_FEED_CHANNEL, + data: { ...event, ...invalid }, + }); + + expect(listener).not.toHaveBeenCalled(); + }, + ); + + it('ignores notifications for other channels', async () => { + const { call, service } = createService(); + const listener = jest.fn(); + service.addListener(listener); + + await service.setActive(true); + getFeedCallback(call)({ + channel: 'social.v1.other', + data: event, + }); + + expect(listener).not.toHaveBeenCalled(); + }); + + it('keeps the feed unsubscribed while the enablement gate is disabled', async () => { + const { call, service } = createService(false); + + await service.setActive(true); + + expect(call).not.toHaveBeenCalled(); + }); + + it('enables the feed by default when no gate is provided', async () => { + const { call, messenger } = createService(); + const service = new SocialRealtimeService({ messenger }); + + await service.setActive(true); + + expect(call).toHaveBeenCalledWith( + 'BackendWebSocketService:subscribe', + expect.objectContaining({ + channels: [SOCIAL_FEED_CHANNEL], + }), + ); + }); + + it('creates an owned subscription when the feed channel already has one', async () => { + const { call, service, unsubscribe } = createService(); + const listener = jest.fn(); + const existingUnsubscribe = jest.fn().mockResolvedValue(undefined); + const existingSubscription = { + unsubscribe: existingUnsubscribe, + } as unknown as WebSocketSubscription; + call.mockImplementation((action: string, _options?: unknown): unknown => { + if (action === 'BackendWebSocketService:connect') { + return Promise.resolve(); + } + + if (action === 'BackendWebSocketService:channelHasSubscription') { + return true; + } + + if (action === 'BackendWebSocketService:getSubscriptionsByChannel') { + return [existingSubscription]; + } + + return Promise.resolve({ unsubscribe }); + }); + + service.addListener(listener); + await service.setActive(true); + getFeedCallback(call)({ + channel: SOCIAL_FEED_CHANNEL, + data: event, + }); + await service.setActive(false); + + expect(call).toHaveBeenCalledWith( + 'BackendWebSocketService:subscribe', + expect.objectContaining({ + channels: [SOCIAL_FEED_CHANNEL], + }), + ); + expect(listener).toHaveBeenCalledWith(event); + expect(existingUnsubscribe).not.toHaveBeenCalled(); + expect(unsubscribe).toHaveBeenCalledTimes(1); + }); + + it('creates a subscription when the backend lookup is empty', async () => { + const { call, service, unsubscribe } = createService(); + call.mockImplementation((action: string, _options?: unknown): unknown => { + if (action === 'BackendWebSocketService:connect') { + return Promise.resolve(); + } + + if (action === 'BackendWebSocketService:channelHasSubscription') { + return true; + } + + if (action === 'BackendWebSocketService:getSubscriptionsByChannel') { + return []; + } + + return Promise.resolve({ unsubscribe }); + }); + + await service.setActive(true); + await service.setActive(false); + + expect(call).toHaveBeenCalledWith( + 'BackendWebSocketService:subscribe', + expect.objectContaining({ + channels: [SOCIAL_FEED_CHANNEL], + }), + ); + expect(unsubscribe).toHaveBeenCalledTimes(1); + }); + + it('avoids creating a second subscription while already active', async () => { + const { call, service } = createService(); + + await service.setActive(true); + await service.setActive(true); + + expect( + call.mock.calls.filter( + ([action]) => action === 'BackendWebSocketService:subscribe', + ), + ).toHaveLength(1); + }); + + it('avoids duplicate subscriptions when connect reports connected synchronously', async () => { + const { call, connectionStateHandler, service, unsubscribe } = + createService(); + let hasReportedConnected = false; + call.mockImplementation((action: string, _options?: unknown): unknown => { + if (action === 'BackendWebSocketService:connect') { + if (!hasReportedConnected) { + hasReportedConnected = true; + connectionStateHandler()?.({ state: WebSocketState.CONNECTED }); + } + return Promise.resolve(); + } + + if (action === 'BackendWebSocketService:channelHasSubscription') { + return false; + } + + return Promise.resolve({ unsubscribe }); + }); + + await service.setActive(true); + await Promise.resolve(); + + expect( + call.mock.calls.filter( + ([action]) => action === 'BackendWebSocketService:subscribe', + ), + ).toHaveLength(1); + }); + + it('ignores an initial connected notification instead of reporting a reconnect', async () => { + const { call, connectionStateHandler, service } = createService(); + const listener = jest.fn(); + service.addReconnectListener(listener); + + await service.setActive(true); + connectionStateHandler()?.({ state: WebSocketState.CONNECTED }); + await Promise.resolve(); + + expect(listener).not.toHaveBeenCalled(); + expect( + call.mock.calls.filter( + ([action]) => action === 'BackendWebSocketService:subscribe', + ), + ).toHaveLength(1); + }); + + it('unsubscribes a subscription that resolves after deactivation', async () => { + const { call, service, unsubscribe } = createService(); + let resolveSubscription: (value: WebSocketSubscription) => void = () => + undefined; + const subscriptionPromise = new Promise( + (resolve): void => { + resolveSubscription = resolve; + }, + ); + call.mockImplementation((action: string, _options?: unknown): unknown => { + if (action === 'BackendWebSocketService:connect') { + return Promise.resolve(); + } + + if (action === 'BackendWebSocketService:channelHasSubscription') { + return false; + } + + if (action === 'BackendWebSocketService:getSubscriptionsByChannel') { + return []; + } + + return subscriptionPromise; + }); + + const activation = service.setActive(true); + await Promise.resolve(); + await Promise.resolve(); + await service.setActive(false); + resolveSubscription({ unsubscribe } as unknown as WebSocketSubscription); + await activation; + + expect(unsubscribe).toHaveBeenCalledTimes(1); + }); + + it('retries subscription setup when reactivation overlaps subscription cleanup', async () => { + const { call, service, unsubscribe } = createService(); + let resolveSubscription: (value: WebSocketSubscription) => void = () => + undefined; + const subscriptionPromise = new Promise( + (resolve): void => { + resolveSubscription = resolve; + }, + ); + let resolveUnsubscribe: () => void = () => undefined; + const unsubscribePromise = new Promise((resolve): void => { + resolveUnsubscribe = resolve; + }); + unsubscribe.mockReturnValue(unsubscribePromise); + call.mockImplementation((action: string, _options?: unknown): unknown => { + if (action === 'BackendWebSocketService:connect') { + return Promise.resolve(); + } + + if (action === 'BackendWebSocketService:channelHasSubscription') { + return false; + } + + if (action === 'BackendWebSocketService:getSubscriptionsByChannel') { + return []; + } + + return subscriptionPromise; + }); + + const activation = service.setActive(true); + await Promise.resolve(); + await Promise.resolve(); + await service.setActive(false); + resolveSubscription({ unsubscribe } as unknown as WebSocketSubscription); + await Promise.resolve(); + + const reactivation = service.setActive(true); + resolveUnsubscribe(); + await activation; + await reactivation; + + expect( + call.mock.calls.filter( + ([action]) => action === 'BackendWebSocketService:subscribe', + ), + ).toHaveLength(2); + }); + + it('swallows subscription errors so feed activation remains best effort', async () => { + const { call, service } = createService(); + const error = new Error('WebSocket unavailable'); + call.mockImplementation((action: string, _options?: unknown): unknown => { + if (action === 'BackendWebSocketService:connect') { + return Promise.reject(error); + } + + return Promise.resolve({ unsubscribe: jest.fn() }); + }); + + await service.setActive(true); + + expect(call).toHaveBeenCalledWith('BackendWebSocketService:connect'); + }); + + it('does not unsubscribe when deactivated without an active subscription', async () => { + const { service, unsubscribe } = createService(); + + await service.setActive(false); + + expect(unsubscribe).not.toHaveBeenCalled(); + }); + + it('removes feed and reconnect listeners', () => { + const { service } = createService(); + const removeFeedListener = service.addListener(jest.fn()); + const removeReconnectListener = service.addReconnectListener(jest.fn()); + + expect(removeFeedListener()).toBe(true); + expect(removeReconnectListener()).toBe(true); + }); + + it('resubscribes after the shared socket reconnects', async () => { + const { call, connectionStateHandler, service } = createService(); + + await service.setActive(true); + connectionStateHandler()?.({ state: WebSocketState.DISCONNECTED }); + connectionStateHandler()?.({ state: WebSocketState.CONNECTED }); + await Promise.resolve(); + await Promise.resolve(); + + expect( + call.mock.calls.filter( + ([action]) => action === 'BackendWebSocketService:subscribe', + ), + ).toHaveLength(2); + }); + + it('notifies active feed listeners after a socket reconnects', async () => { + const { connectionStateHandler, service } = createService(); + const listener = jest.fn(); + service.addReconnectListener(listener); + + await service.setActive(true); + connectionStateHandler()?.({ state: WebSocketState.DISCONNECTED }); + connectionStateHandler()?.({ state: WebSocketState.CONNECTED }); + + expect(listener).toHaveBeenCalledTimes(1); + }); + + it('unsubscribes when the feed leaves the active tab', async () => { + const { service, unsubscribe } = createService(); + + await service.setActive(true); + await service.setActive(false); + + expect(unsubscribe).toHaveBeenCalledTimes(1); + }); +}); diff --git a/packages/social-controllers/src/SocialRealtimeService.ts b/packages/social-controllers/src/SocialRealtimeService.ts new file mode 100644 index 00000000000..d0771ee2388 --- /dev/null +++ b/packages/social-controllers/src/SocialRealtimeService.ts @@ -0,0 +1,199 @@ +import { WebSocketState } from '@metamask/core-backend'; +import type { + BackendWebSocketServiceMessenger, + ServerNotificationMessage, + WebSocketSubscription, +} from '@metamask/core-backend'; + +import type { FeedItem } from './social-types.js'; + +export const SOCIAL_FEED_CHANNEL = 'social.v1.feed.all' as const; +export const SOCIAL_FEED_CHANNEL_TYPE = 'social.v1' as const; + +export type SocialFeedEvent = { + version: 1; + kind: 'feed-item'; + eventId: string; + feedItemId: string; + revision: number; + occurredAt: string; + data: FeedItem; +}; + +export type SocialFeedEventListener = (event: SocialFeedEvent) => void; +export type SocialFeedReconnectListener = () => void; + +export type SocialRealtimeServiceOptions = { + messenger: BackendWebSocketServiceMessenger; + isEnabled?: () => boolean; +}; + +const isRecord = (value: unknown): value is Record => + typeof value === 'object' && value !== null; + +const parseSocialFeedEvent = ( + data: Record, +): SocialFeedEvent | undefined => { + if ( + data.version !== 1 || + data.kind !== 'feed-item' || + typeof data.eventId !== 'string' || + typeof data.feedItemId !== 'string' || + typeof data.revision !== 'number' || + typeof data.occurredAt !== 'string' || + !isRecord(data.data) + ) { + return undefined; + } + + return data as unknown as SocialFeedEvent; +}; + +/** + * Social realtime subscription service built on the shared backend WebSocket. + * + * The POC subscribes to the global feed channel. Additional Social channels + * can be added here without creating another WebSocket connection. + */ +export class SocialRealtimeService { + readonly name = 'SocialRealtimeService' as const; + + readonly #messenger: BackendWebSocketServiceMessenger; + + readonly #listeners = new Set(); + + readonly #reconnectListeners = new Set(); + + readonly #isEnabled: () => boolean; + + #active = false; + + #subscription: WebSocketSubscription | undefined; + + #subscriptionPromise: Promise | undefined; + + #wasDisconnected = false; + + constructor({ + messenger, + isEnabled = (): boolean => true, + }: SocialRealtimeServiceOptions) { + this.#messenger = messenger; + this.#isEnabled = isEnabled; + + this.#messenger.subscribe( + 'BackendWebSocketService:connectionStateChanged', + ({ state }: { state: WebSocketState }): void => { + if (state === WebSocketState.DISCONNECTED) { + this.#subscription = undefined; + this.#wasDisconnected = true; + } + + if (state === WebSocketState.CONNECTED && this.#active) { + const isReconnect = this.#wasDisconnected; + this.#wasDisconnected = false; + // Subscription setup handles its own errors and must not block reconnect handling. + // eslint-disable-next-line no-void + void this.#subscribeToFeed(); + + if (isReconnect) { + this.#reconnectListeners.forEach((listener) => listener()); + } + } + }, + ); + } + + addListener(listener: SocialFeedEventListener): () => void { + this.#listeners.add(listener); + return () => this.#listeners.delete(listener); + } + + addReconnectListener(listener: SocialFeedReconnectListener): () => void { + this.#reconnectListeners.add(listener); + return () => this.#reconnectListeners.delete(listener); + } + + async setActive(active: boolean): Promise { + this.#active = active && this.#isEnabled(); + + if (!this.#active) { + await this.#unsubscribeFromFeed(); + return; + } + + await this.#subscribeToFeed(); + } + + async #subscribeToFeed(): Promise { + if (!this.#active || this.#subscription) { + return; + } + + while (this.#subscriptionPromise) { + await this.#subscriptionPromise; + + if (!this.#active || this.#subscription) { + return; + } + } + + const subscriptionPromise = Promise.resolve().then( + async (): Promise => { + try { + await this.#messenger.call('BackendWebSocketService:connect'); + + // Each service needs its own callback and subscription lifecycle. + const subscription = await this.#messenger.call( + 'BackendWebSocketService:subscribe', + { + channels: [SOCIAL_FEED_CHANNEL], + channelType: SOCIAL_FEED_CHANNEL_TYPE, + callback: (notification: ServerNotificationMessage) => { + this.#handleNotification(notification); + }, + }, + ); + + if (!this.#active) { + await subscription.unsubscribe(); + return; + } + + this.#subscription = subscription; + } catch { + this.#subscription = undefined; + } + }, + ); + this.#subscriptionPromise = subscriptionPromise; + + try { + await subscriptionPromise; + } finally { + this.#subscriptionPromise = undefined; + } + } + + async #unsubscribeFromFeed(): Promise { + const subscription = this.#subscription; + this.#subscription = undefined; + + if (subscription) { + await subscription.unsubscribe(); + } + } + + #handleNotification(notification: ServerNotificationMessage): void { + if (notification.channel !== SOCIAL_FEED_CHANNEL) { + return; + } + + const event = parseSocialFeedEvent(notification.data); + if (!event) { + return; + } + + this.#listeners.forEach((listener) => listener(event)); + } +} diff --git a/packages/social-controllers/src/index.test.ts b/packages/social-controllers/src/index.test.ts index 21e4fc8b486..a8a01e14ce8 100644 --- a/packages/social-controllers/src/index.test.ts +++ b/packages/social-controllers/src/index.test.ts @@ -1,4 +1,9 @@ -import { SocialController, SocialService, socialServiceName } from './index.js'; +import { + SocialController, + SocialRealtimeService, + SocialService, + socialServiceName, +} from './index.js'; describe('social-controllers exports', () => { it('exports SocialController class', () => { @@ -9,6 +14,10 @@ describe('social-controllers exports', () => { expect(SocialService).toBeDefined(); }); + it('exports SocialRealtimeService class', () => { + expect(SocialRealtimeService).toBeDefined(); + }); + it('exports serviceName', () => { expect(socialServiceName).toBe('SocialService'); }); diff --git a/packages/social-controllers/src/index.ts b/packages/social-controllers/src/index.ts index 60a7ae708c9..213c9222cd4 100644 --- a/packages/social-controllers/src/index.ts +++ b/packages/social-controllers/src/index.ts @@ -49,6 +49,17 @@ export type { } from './SocialService-method-action-types.js'; export { TRADER_RANKING_TAGS, TradeStruct } from './social-types.js'; +export { + SOCIAL_FEED_CHANNEL, + SOCIAL_FEED_CHANNEL_TYPE, + SocialRealtimeService, +} from './SocialRealtimeService.js'; +export type { + SocialFeedEvent, + SocialFeedEventListener, + SocialFeedReconnectListener, + SocialRealtimeServiceOptions, +} from './SocialRealtimeService.js'; export type { AuthorComment, CommentEngagement, diff --git a/packages/social-controllers/tsconfig.build.json b/packages/social-controllers/tsconfig.build.json index 9325a144097..03b2b1ef87e 100644 --- a/packages/social-controllers/tsconfig.build.json +++ b/packages/social-controllers/tsconfig.build.json @@ -14,6 +14,9 @@ { "path": "../controller-utils/tsconfig.build.json" }, + { + "path": "../core-backend/tsconfig.build.json" + }, { "path": "../messenger/tsconfig.build.json" }, diff --git a/packages/social-controllers/tsconfig.json b/packages/social-controllers/tsconfig.json index faef108f266..b94e511e23a 100644 --- a/packages/social-controllers/tsconfig.json +++ b/packages/social-controllers/tsconfig.json @@ -10,6 +10,9 @@ { "path": "../controller-utils" }, + { + "path": "../core-backend" + }, { "path": "../messenger" }, diff --git a/packages/social-controllers/tsconfig.lint.json b/packages/social-controllers/tsconfig.lint.json index cf316f6a069..e49684e691d 100644 --- a/packages/social-controllers/tsconfig.lint.json +++ b/packages/social-controllers/tsconfig.lint.json @@ -17,6 +17,9 @@ { "path": "../controller-utils/tsconfig.lint.json" }, + { + "path": "../core-backend/tsconfig.lint.json" + }, { "path": "../profile-sync-controller/tsconfig.lint.json" } diff --git a/yarn.lock b/yarn.lock index b837acb372a..f88bfde113c 100644 --- a/yarn.lock +++ b/yarn.lock @@ -9057,6 +9057,7 @@ __metadata: "@metamask/base-controller": "npm:^10.0.0" "@metamask/base-data-service": "npm:^2.1.0" "@metamask/controller-utils": "npm:^13.0.0" + "@metamask/core-backend": "npm:^11.0.0" "@metamask/messenger": "npm:^3.0.0" "@metamask/profile-sync-controller": "npm:^33.0.0" "@metamask/superstruct": "npm:^3.4.1"