From 3e74bc076928b8cef0e23b6628d27e7cc58cb138 Mon Sep 17 00:00:00 2001 From: Yoav Farhi Date: Fri, 15 May 2026 14:26:28 +0300 Subject: [PATCH 1/2] Add entity subscription dedupe and caps --- src/client.ts | 17 +- src/client.types.ts | 9 +- src/index.ts | 1 + src/modules/entities.ts | 403 +++++++++++++++++++++++--- src/modules/entities.types.ts | 37 +++ src/utils/socket-utils.ts | 3 +- tests/unit/entities-subscribe.test.ts | 273 ++++++++++++++++- tests/unit/socket-utils.test.ts | 71 +++++ 8 files changed, 764 insertions(+), 50 deletions(-) create mode 100644 tests/unit/socket-utils.test.ts diff --git a/src/client.ts b/src/client.ts index a028f337..8e16bfaa 100644 --- a/src/client.ts +++ b/src/client.ts @@ -162,11 +162,20 @@ export function createClient(config: CreateClientConfig): Base44Client { } } + const userAnalyticsModule = createAnalyticsModule({ + axiosClient, + serverUrl, + appId, + userAuthModule, + }); + const userModules = { entities: createEntitiesModule({ axios: axiosClient, appId, getSocket, + subscriptionOptions: options?.entitySubscriptions, + trackSubscriptionEvent: userAnalyticsModule.track, }), integrations: createIntegrationsModule(axiosClient, appId), connectors: createUserConnectorsModule(axiosClient, appId), @@ -192,12 +201,7 @@ export function createClient(config: CreateClientConfig): Base44Client { }), appLogs: createAppLogsModule(axiosClient, appId), users: createUsersModule(axiosClient, appId), - analytics: createAnalyticsModule({ - axiosClient, - serverUrl, - appId, - userAuthModule, - }), + analytics: userAnalyticsModule, cleanup: () => { userModules.analytics.cleanup(); if (socket) { @@ -211,6 +215,7 @@ export function createClient(config: CreateClientConfig): Base44Client { axios: serviceRoleAxiosClient, appId, getSocket, + subscriptionOptions: options?.entitySubscriptions, }), integrations: createIntegrationsModule(serviceRoleAxiosClient, appId), sso: createSsoModule(serviceRoleAxiosClient, appId), diff --git a/src/client.types.ts b/src/client.types.ts index 6b4c9c57..6f54f413 100644 --- a/src/client.types.ts +++ b/src/client.types.ts @@ -1,4 +1,7 @@ -import type { EntitiesModule } from "./modules/entities.types.js"; +import type { + EntitiesModule, + EntitySubscriptionOptions, +} from "./modules/entities.types.js"; import type { IntegrationsModule } from "./modules/integrations.types.js"; import type { AuthModule } from "./modules/auth.types.js"; import type { SsoModule } from "./modules/sso.types.js"; @@ -19,6 +22,10 @@ export interface CreateClientOptions { * Optional error handler that will be called whenever an API error occurs. */ onError?: (error: Error) => void; + /** + * Client-side controls for realtime entity subscriptions. + */ + entitySubscriptions?: EntitySubscriptionOptions; } /** diff --git a/src/index.ts b/src/index.ts index bc531d89..f1d29d4b 100644 --- a/src/index.ts +++ b/src/index.ts @@ -42,6 +42,7 @@ export type { EntityFilterValue, EntityHandler, EntityRecord, + EntitySubscriptionOptions, EntityTypeRegistry, ImportResult, RealtimeEventType, diff --git a/src/modules/entities.ts b/src/modules/entities.ts index 1eaaf287..dc1523c5 100644 --- a/src/modules/entities.ts +++ b/src/modules/entities.ts @@ -5,6 +5,7 @@ import { EntitiesModule, EntityFilterQuery, EntityHandler, + EntitySubscriptionOptions, ImportResult, RealtimeCallback, RealtimeEvent, @@ -12,6 +13,7 @@ import { SortField, UpdateManyResult, } from "./entities.types"; +import type { TrackEventParams } from "./analytics.types"; import { RoomsSocket } from "../utils/socket-utils.js"; /** @@ -22,6 +24,41 @@ export interface EntitiesModuleConfig { axios: AxiosInstance; appId: string; getSocket: () => ReturnType; + subscriptionOptions?: EntitySubscriptionOptions; + trackSubscriptionEvent?: (params: TrackEventParams) => void; +} + +const DEFAULT_MAX_ACTIVE_ENTITY_SUBSCRIPTIONS = 100; +const DEFAULT_SUBSCRIPTION_CHURN_WARNING_THRESHOLD = 20; +const DEFAULT_SUBSCRIPTION_CHURN_WINDOW_MS = 60_000; +const DEFAULT_EMPTY_ROOM_GRACE_MS = 1_000; +const ENTITY_SUBSCRIPTION_WARNING_EVENT_NAME = + "__entity_subscription_warning__"; + +type SubscriptionChurnAction = "subscribe" | "unsubscribe"; + +type NormalizedEntitySubscriptionOptions = Required; + +interface EntitySubscriptionManager { + subscribe(entityName: string, callback: RealtimeCallback): () => void; +} + +interface EntitySubscriptionState { + room: string; + entityName: string; + callbacks: Map>; + unsubscribeFromRoom: () => void; + closeTimer: ReturnType | null; +} + +interface SubscriptionChurnRecord { + action: SubscriptionChurnAction; + timestamp: number; +} + +interface SubscriptionChurnState { + events: SubscriptionChurnRecord[]; + lastWarningAt: number | null; } /** @@ -35,6 +72,14 @@ export function createEntitiesModule( config: EntitiesModuleConfig ): EntitiesModule { const { axios, appId, getSocket } = config; + const entityHandlers = new Map>(); + const subscriptionManager = createEntitySubscriptionManager({ + appId, + getSocket, + options: config.subscriptionOptions, + trackSubscriptionEvent: config.trackSubscriptionEvent, + }); + // Using Proxy to dynamically handle entity names return new Proxy( {}, @@ -49,8 +94,20 @@ export function createEntitiesModule( return undefined; } + const cachedHandler = entityHandlers.get(entityName); + if (cachedHandler) { + return cachedHandler; + } + // Create entity handler - return createEntityHandler(axios, appId, entityName, getSocket); + const handler = createEntityHandler( + axios, + appId, + entityName, + subscriptionManager + ); + entityHandlers.set(entityName, handler); + return handler; }, } ) as EntitiesModule; @@ -75,13 +132,316 @@ function parseRealtimeMessage(dataStr: string): RealtimeEvent | null } } +function normalizeEntitySubscriptionOptions( + options?: EntitySubscriptionOptions +): NormalizedEntitySubscriptionOptions { + return { + maxActiveSubscriptions: normalizePositiveInteger( + options?.maxActiveSubscriptions, + DEFAULT_MAX_ACTIVE_ENTITY_SUBSCRIPTIONS + ), + churnWarningThreshold: normalizePositiveInteger( + options?.churnWarningThreshold, + DEFAULT_SUBSCRIPTION_CHURN_WARNING_THRESHOLD + ), + churnWindowMs: normalizePositiveInteger( + options?.churnWindowMs, + DEFAULT_SUBSCRIPTION_CHURN_WINDOW_MS + ), + emptyRoomGraceMs: normalizeNonNegativeInteger( + options?.emptyRoomGraceMs, + DEFAULT_EMPTY_ROOM_GRACE_MS + ), + }; +} + +function normalizePositiveInteger( + value: number | undefined, + fallback: number +): number { + if (typeof value !== "number" || !Number.isFinite(value) || value <= 0) { + return fallback; + } + return Math.floor(value); +} + +function normalizeNonNegativeInteger( + value: number | undefined, + fallback: number +): number { + if (typeof value !== "number" || !Number.isFinite(value) || value < 0) { + return fallback; + } + return Math.floor(value); +} + +function createEntitySubscriptionManager({ + appId, + getSocket, + options, + trackSubscriptionEvent, +}: { + appId: string; + getSocket: () => ReturnType; + options?: EntitySubscriptionOptions; + trackSubscriptionEvent?: (params: TrackEventParams) => void; +}): EntitySubscriptionManager { + const normalizedOptions = normalizeEntitySubscriptionOptions(options); + const activeSubscriptions = new Map(); + const churnByRoom = new Map(); + const roomsWarnedForCap = new Set(); + let nextCallbackId = 1; + + function makeRoom(entityName: string) { + return `entities:${appId}:${entityName}`; + } + + function emitWarning( + message: string, + properties: NonNullable + ) { + console.warn(message); + + try { + trackSubscriptionEvent?.({ + eventName: ENTITY_SUBSCRIPTION_WARNING_EVENT_NAME, + properties, + }); + } catch { + // Diagnostics should never break application code. + } + } + + function recordSubscriptionActivity( + room: string, + entityName: string, + action: SubscriptionChurnAction + ) { + const now = Date.now(); + const churnState = + churnByRoom.get(room) ?? + ({ + events: [], + lastWarningAt: null, + } satisfies SubscriptionChurnState); + + churnState.events = churnState.events.filter( + (event) => now - event.timestamp <= normalizedOptions.churnWindowMs + ); + churnState.events.push({ action, timestamp: now }); + churnByRoom.set(room, churnState); + + const subscribeCount = churnState.events.filter( + (event) => event.action === "subscribe" + ).length; + const unsubscribeCount = churnState.events.length - subscribeCount; + + if ( + churnState.events.length < normalizedOptions.churnWarningThreshold || + subscribeCount === 0 || + unsubscribeCount === 0 || + (churnState.lastWarningAt !== null && + now - churnState.lastWarningAt < normalizedOptions.churnWindowMs) + ) { + return; + } + + churnState.lastWarningAt = now; + emitWarning( + `[Base44 SDK] entities.${entityName}.subscribe() is being created and cleaned up repeatedly ` + + `(${churnState.events.length} subscribe/unsubscribe operations in ` + + `${normalizedOptions.churnWindowMs}ms). Keep realtime subscriptions in a stable lifecycle to avoid socket churn.`, + { + reason: "subscription_churn", + app_id: appId, + entity: entityName, + room, + action, + activity_count: churnState.events.length, + subscribe_count: subscribeCount, + unsubscribe_count: unsubscribeCount, + churn_window_ms: normalizedOptions.churnWindowMs, + empty_room_grace_ms: normalizedOptions.emptyRoomGraceMs, + churn_warning_threshold: + normalizedOptions.churnWarningThreshold, + } + ); + } + + function warnForSubscriptionCap(room: string, entityName: string) { + if (roomsWarnedForCap.has(room)) { + return; + } + + roomsWarnedForCap.add(room); + emitWarning( + `[Base44 SDK] Realtime entity subscription cap reached ` + + `(${normalizedOptions.maxActiveSubscriptions} active entities per SDK client/tab). ` + + `Skipping entities.${entityName}.subscribe(). Unsubscribe from unused entity subscriptions before subscribing to more entities.`, + { + reason: "active_subscription_cap", + app_id: appId, + entity: entityName, + room, + active_subscription_count: activeSubscriptions.size, + max_active_subscriptions: + normalizedOptions.maxActiveSubscriptions, + } + ); + } + + function closeRoomSubscription(state: EntitySubscriptionState) { + clearPendingClose(state); + state.unsubscribeFromRoom(); + activeSubscriptions.delete(state.room); + + if ( + activeSubscriptions.size < normalizedOptions.maxActiveSubscriptions + ) { + roomsWarnedForCap.clear(); + } + } + + function clearPendingClose(state: EntitySubscriptionState) { + if (!state.closeTimer) { + return; + } + + clearTimeout(state.closeTimer); + state.closeTimer = null; + } + + function scheduleRoomClose(state: EntitySubscriptionState) { + if (state.closeTimer) { + return; + } + + if (normalizedOptions.emptyRoomGraceMs === 0) { + closeRoomSubscription(state); + return; + } + + const closeTimer = setTimeout(() => { + state.closeTimer = null; + + if ( + state.callbacks.size === 0 && + activeSubscriptions.get(state.room) === state + ) { + closeRoomSubscription(state); + } + }, normalizedOptions.emptyRoomGraceMs); + + closeTimer.unref?.(); + state.closeTimer = closeTimer; + } + + function dispatchRealtimeMessage( + state: EntitySubscriptionState, + dataStr: string + ) { + if (state.callbacks.size === 0) { + return; + } + + const event = parseRealtimeMessage(dataStr); + if (!event) { + return; + } + + // Server signals oversize broadcasts with `_oversize: true` on + // `data`. The wire payload was slimmed to fit under the realtime + // transport cap, so big string fields arrive as empty strings (or + // the whole record collapses to a stub). Surface this to the + // developer console so they know to fetch the full record on + // demand (e.g. a follow-up entities.X.get(id) call) instead of + // rendering the slimmed payload directly. Skip on delete events + // — the record no longer exists. + if (event.type !== "delete" && (event.data as any)?._oversize) { + console.error( + `[Base44 SDK] Realtime broadcast for ${state.entityName}#${event.id} was oversize and got slimmed for transport. ` + + `Fields >10 KB are empty and the rest of the record may be a stub. ` + + `Call \`entities.${state.entityName}.get("${event.id}")\` to fetch the full record.` + ); + } + + Array.from(state.callbacks.values()).forEach((callback) => { + try { + callback(event); + } catch (error) { + console.error("[Base44 SDK] Subscription callback error:", error); + } + }); + } + + function openRoomSubscription(entityName: string, room: string) { + const state: EntitySubscriptionState = { + room, + entityName, + callbacks: new Map(), + unsubscribeFromRoom: () => {}, + closeTimer: null, + }; + const socket = getSocket(); + + state.unsubscribeFromRoom = socket.subscribeToRoom(room, { + update_model: (msg) => { + dispatchRealtimeMessage(state, msg.data); + }, + }); + activeSubscriptions.set(room, state); + return state; + } + + return { + subscribe(entityName: string, callback: RealtimeCallback) { + const room = makeRoom(entityName); + recordSubscriptionActivity(room, entityName, "subscribe"); + + let state = activeSubscriptions.get(room); + if (!state) { + if ( + activeSubscriptions.size >= + normalizedOptions.maxActiveSubscriptions + ) { + warnForSubscriptionCap(room, entityName); + return () => {}; + } + state = openRoomSubscription(entityName, room); + } else { + clearPendingClose(state); + } + + const callbackId = nextCallbackId++; + state.callbacks.set(callbackId, callback as RealtimeCallback); + + let unsubscribed = false; + return () => { + if (unsubscribed) { + return; + } + unsubscribed = true; + recordSubscriptionActivity(room, entityName, "unsubscribe"); + state.callbacks.delete(callbackId); + + if ( + state.callbacks.size === 0 && + activeSubscriptions.get(room) === state + ) { + scheduleRoomClose(state); + } + }; + }, + }; +} + /** * Creates a handler for a specific entity. * * @param axios - Axios instance * @param appId - Application ID * @param entityName - Entity name - * @param getSocket - Function to get the socket instance + * @param subscriptionManager - Shared realtime subscription manager * @returns Entity handler with CRUD methods * @internal */ @@ -89,7 +449,7 @@ function createEntityHandler( axios: AxiosInstance, appId: string, entityName: string, - getSocket: () => ReturnType + subscriptionManager: EntitySubscriptionManager ): EntityHandler { const baseURL = `/apps/${appId}/entities/${entityName}`; @@ -186,42 +546,7 @@ function createEntityHandler( // Subscribe to realtime updates subscribe(callback: RealtimeCallback): () => void { - const room = `entities:${appId}:${entityName}`; - - // Get the socket and subscribe to the room - const socket = getSocket(); - const unsubscribe = socket.subscribeToRoom(room, { - update_model: (msg) => { - const event = parseRealtimeMessage(msg.data); - if (!event) { - return; - } - - // Server signals oversize broadcasts with `_oversize: true` on - // `data`. The wire payload was slimmed to fit under the realtime - // transport cap, so big string fields arrive as empty strings (or - // the whole record collapses to a stub). Surface this to the - // developer console so they know to fetch the full record on - // demand (e.g. a follow-up entities.X.get(id) call) instead of - // rendering the slimmed payload directly. Skip on delete events - // — the record no longer exists. - if (event.type !== "delete" && (event.data as any)?._oversize) { - console.error( - `[Base44 SDK] Realtime broadcast for ${entityName}#${event.id} was oversize and got slimmed for transport. ` + - `Fields >10 KB are empty and the rest of the record may be a stub. ` + - `Call \`entities.${entityName}.get("${event.id}")\` to fetch the full record.` - ); - } - - try { - callback(event); - } catch (error) { - console.error("[Base44 SDK] Subscription callback error:", error); - } - }, - }); - - return unsubscribe; + return subscriptionManager.subscribe(entityName, callback); }, }; } diff --git a/src/modules/entities.types.ts b/src/modules/entities.types.ts index c5854dd3..bab7a86a 100644 --- a/src/modules/entities.types.ts +++ b/src/modules/entities.types.ts @@ -26,6 +26,43 @@ export interface RealtimeEvent { */ export type RealtimeCallback = (event: RealtimeEvent) => void; +/** + * Client-side controls for realtime entity subscriptions. + */ +export interface EntitySubscriptionOptions { + /** + * Maximum number of distinct active entity realtime subscriptions allowed + * for one SDK client instance. Repeated subscriptions to the same entity + * share one active entity subscription and do not count again. + * + * @defaultValue `100` + */ + maxActiveSubscriptions?: number; + /** + * Number of subscribe/unsubscribe operations for the same entity within + * `churnWindowMs` before the SDK logs a warning and emits diagnostic + * telemetry. + * + * @defaultValue `20` + */ + churnWarningThreshold?: number; + /** + * Time window, in milliseconds, used to detect repeated subscribe/unsubscribe + * churn for one entity. + * + * @defaultValue `60000` + */ + churnWindowMs?: number; + /** + * Grace period, in milliseconds, before the SDK leaves an entity realtime + * room after its last local callback unsubscribes. A new subscription to + * the same entity during this window reuses the existing room membership. + * + * @defaultValue `1000` + */ + emptyRoomGraceMs?: number; +} + /** * Result returned when deleting a single entity. */ diff --git a/src/utils/socket-utils.ts b/src/utils/socket-utils.ts index 2f731311..14598b15 100644 --- a/src/utils/socket-utils.ts +++ b/src/utils/socket-utils.ts @@ -138,7 +138,7 @@ export function RoomsSocket({ config }: { config: RoomsSocketConfig }) { } function getListeners(room: string) { - return roomsToListeners[room]; + return roomsToListeners[room] ?? []; } const subscribeToRoom = ( @@ -158,6 +158,7 @@ export function RoomsSocket({ config }: { config: RoomsSocketConfig }) { []; if (roomsToListeners[room].length === 0) { leaveRoom(room); + delete roomsToListeners[room]; } }; }; diff --git a/tests/unit/entities-subscribe.test.ts b/tests/unit/entities-subscribe.test.ts index 7ac3d905..482cd959 100644 --- a/tests/unit/entities-subscribe.test.ts +++ b/tests/unit/entities-subscribe.test.ts @@ -7,19 +7,22 @@ describe("Entities Module - subscribe()", () => { // Helper to create a mock socket function createMockSocket() { const listeners: Record = {}; + const unsubscribes: Record> = {}; return { subscribeToRoom: vi.fn((room: string, handlers: any) => { listeners[room] = handlers; - // Return unsubscribe function - return () => { + const unsubscribe = vi.fn(() => { delete listeners[room]; - }; + }); + unsubscribes[room] = unsubscribe; + return unsubscribe; }), // Helper to simulate incoming messages _simulateMessage: (room: string, msg: any) => { listeners[room]?.update_model?.(msg); }, _getListeners: () => listeners, + _getUnsubscribe: (room: string) => unsubscribes[room], }; } @@ -177,6 +180,270 @@ describe("Entities Module - subscribe()", () => { expect(callback).toHaveBeenCalledTimes(1); }); + test("subscribe() should fan out callbacks through one socket room subscription", async () => { + vi.useFakeTimers(); + const mockSocket = createMockSocket(); + const mockAxios = createMockAxios(); + + try { + const entities = createEntitiesModule({ + axios: mockAxios as any, + appId, + getSocket: () => mockSocket as any, + }); + + const firstCallback = vi.fn(); + const secondCallback = vi.fn(); + const firstUnsubscribe = entities.Todo.subscribe(firstCallback); + const secondUnsubscribe = entities.Todo.subscribe(secondCallback); + const room = `entities:${appId}:Todo`; + const roomUnsubscribe = mockSocket._getUnsubscribe(room); + + expect(mockSocket.subscribeToRoom).toHaveBeenCalledTimes(1); + + mockSocket._simulateMessage(room, { + room, + data: JSON.stringify({ + type: "create", + data: { id: "1" }, + id: "1", + timestamp: "2024-01-01T00:00:00.000Z", + }), + }); + + expect(firstCallback).toHaveBeenCalledTimes(1); + expect(secondCallback).toHaveBeenCalledTimes(1); + + firstUnsubscribe(); + + mockSocket._simulateMessage(room, { + room, + data: JSON.stringify({ + type: "update", + data: { id: "1" }, + id: "1", + timestamp: "2024-01-01T00:00:00.000Z", + }), + }); + + expect(firstCallback).toHaveBeenCalledTimes(1); + expect(secondCallback).toHaveBeenCalledTimes(2); + expect(roomUnsubscribe).not.toHaveBeenCalled(); + + secondUnsubscribe(); + + expect(roomUnsubscribe).not.toHaveBeenCalled(); + + await vi.advanceTimersByTimeAsync(1_000); + + expect(roomUnsubscribe).toHaveBeenCalledTimes(1); + } finally { + vi.useRealTimers(); + } + }); + + test("subscribe() should cancel the empty-room leave when resubscribed during grace", async () => { + vi.useFakeTimers(); + const mockSocket = createMockSocket(); + const mockAxios = createMockAxios(); + + try { + const entities = createEntitiesModule({ + axios: mockAxios as any, + appId, + getSocket: () => mockSocket as any, + subscriptionOptions: { emptyRoomGraceMs: 1_000 }, + }); + const room = `entities:${appId}:Todo`; + const firstUnsubscribe = entities.Todo.subscribe(vi.fn()); + const roomUnsubscribe = mockSocket._getUnsubscribe(room); + + firstUnsubscribe(); + + expect(roomUnsubscribe).not.toHaveBeenCalled(); + + const secondCallback = vi.fn(); + const secondUnsubscribe = entities.Todo.subscribe(secondCallback); + + await vi.advanceTimersByTimeAsync(1_000); + + expect(mockSocket.subscribeToRoom).toHaveBeenCalledTimes(1); + expect(roomUnsubscribe).not.toHaveBeenCalled(); + + mockSocket._simulateMessage(room, { + room, + data: JSON.stringify({ + type: "update", + data: { id: "1" }, + id: "1", + timestamp: "2024-01-01T00:00:00.000Z", + }), + }); + + expect(secondCallback).toHaveBeenCalledTimes(1); + + secondUnsubscribe(); + await vi.advanceTimersByTimeAsync(1_000); + + expect(roomUnsubscribe).toHaveBeenCalledTimes(1); + } finally { + vi.useRealTimers(); + } + }); + + test("subscribe() should leave an empty room after the grace period expires", async () => { + vi.useFakeTimers(); + const mockSocket = createMockSocket(); + const mockAxios = createMockAxios(); + + try { + const entities = createEntitiesModule({ + axios: mockAxios as any, + appId, + getSocket: () => mockSocket as any, + subscriptionOptions: { emptyRoomGraceMs: 1_000 }, + }); + const room = `entities:${appId}:Todo`; + const unsubscribe = entities.Todo.subscribe(vi.fn()); + const roomUnsubscribe = mockSocket._getUnsubscribe(room); + + unsubscribe(); + await vi.advanceTimersByTimeAsync(999); + + expect(roomUnsubscribe).not.toHaveBeenCalled(); + + await vi.advanceTimersByTimeAsync(1); + + expect(roomUnsubscribe).toHaveBeenCalledTimes(1); + } finally { + vi.useRealTimers(); + } + }); + + test("subscribe() should cap distinct active entity subscriptions", () => { + const mockSocket = createMockSocket(); + const mockAxios = createMockAxios(); + const trackSubscriptionEvent = vi.fn(); + const warnSpy = vi.spyOn(console, "warn").mockImplementation(() => {}); + + const entities = createEntitiesModule({ + axios: mockAxios as any, + appId, + getSocket: () => mockSocket as any, + subscriptionOptions: { maxActiveSubscriptions: 1 }, + trackSubscriptionEvent, + }); + + const todoCallback = vi.fn(); + const userCallback = vi.fn(); + const unsubscribeTodo = entities.Todo.subscribe(todoCallback); + const unsubscribeUser = entities.User.subscribe(userCallback); + + expect(mockSocket.subscribeToRoom).toHaveBeenCalledTimes(1); + expect(mockSocket.subscribeToRoom).toHaveBeenCalledWith( + `entities:${appId}:Todo`, + expect.any(Object) + ); + expect(warnSpy).toHaveBeenCalledWith( + expect.stringContaining("Realtime entity subscription cap reached") + ); + expect(trackSubscriptionEvent).toHaveBeenCalledWith({ + eventName: "__entity_subscription_warning__", + properties: expect.objectContaining({ + reason: "active_subscription_cap", + entity: "User", + active_subscription_count: 1, + max_active_subscriptions: 1, + }), + }); + + mockSocket._simulateMessage(`entities:${appId}:User`, { + room: `entities:${appId}:User`, + data: JSON.stringify({ + type: "create", + data: { id: "blocked" }, + id: "blocked", + timestamp: "2024-01-01T00:00:00.000Z", + }), + }); + + expect(userCallback).not.toHaveBeenCalled(); + + unsubscribeUser(); + unsubscribeTodo(); + warnSpy.mockRestore(); + }); + + test("subscribe() should warn and emit telemetry on repeated subscription churn", () => { + const mockSocket = createMockSocket(); + const mockAxios = createMockAxios(); + const trackSubscriptionEvent = vi.fn(); + const warnSpy = vi.spyOn(console, "warn").mockImplementation(() => {}); + + const entities = createEntitiesModule({ + axios: mockAxios as any, + appId, + getSocket: () => mockSocket as any, + subscriptionOptions: { + churnWarningThreshold: 4, + churnWindowMs: 60_000, + }, + trackSubscriptionEvent, + }); + + const firstUnsubscribe = entities.Todo.subscribe(vi.fn()); + firstUnsubscribe(); + const secondUnsubscribe = entities.Todo.subscribe(vi.fn()); + secondUnsubscribe(); + + expect(warnSpy).toHaveBeenCalledWith( + expect.stringContaining("created and cleaned up repeatedly") + ); + expect(trackSubscriptionEvent).toHaveBeenCalledWith({ + eventName: "__entity_subscription_warning__", + properties: expect.objectContaining({ + reason: "subscription_churn", + entity: "Todo", + activity_count: 4, + subscribe_count: 2, + unsubscribe_count: 2, + churn_window_ms: 60000, + churn_warning_threshold: 4, + }), + }); + + warnSpy.mockRestore(); + }); + + test("subscribe() should not warn when many callbacks fan out without unsubscribe churn", () => { + const mockSocket = createMockSocket(); + const mockAxios = createMockAxios(); + const trackSubscriptionEvent = vi.fn(); + const warnSpy = vi.spyOn(console, "warn").mockImplementation(() => {}); + + const entities = createEntitiesModule({ + axios: mockAxios as any, + appId, + getSocket: () => mockSocket as any, + subscriptionOptions: { + churnWarningThreshold: 4, + churnWindowMs: 60_000, + }, + trackSubscriptionEvent, + }); + + entities.Todo.subscribe(vi.fn()); + entities.Todo.subscribe(vi.fn()); + entities.Todo.subscribe(vi.fn()); + entities.Todo.subscribe(vi.fn()); + + expect(mockSocket.subscribeToRoom).toHaveBeenCalledTimes(1); + expect(warnSpy).not.toHaveBeenCalled(); + expect(trackSubscriptionEvent).not.toHaveBeenCalled(); + + warnSpy.mockRestore(); + }); + test("subscribe() should not call callback for invalid JSON messages", () => { const mockSocket = createMockSocket(); const mockAxios = createMockAxios(); diff --git a/tests/unit/socket-utils.test.ts b/tests/unit/socket-utils.test.ts new file mode 100644 index 00000000..116183b6 --- /dev/null +++ b/tests/unit/socket-utils.test.ts @@ -0,0 +1,71 @@ +import { beforeEach, describe, expect, test, vi } from "vitest"; +import { RoomsSocket } from "../../src/utils/socket-utils.ts"; + +const socketMock = vi.hoisted(() => ({ + disconnect: vi.fn(), + emit: vi.fn(), + listeners: {} as Record void>, + on: vi.fn(), +})); + +vi.mock("socket.io-client", () => ({ + io: vi.fn(() => ({ + id: "socket-id", + disconnect: socketMock.disconnect, + emit: socketMock.emit, + on: socketMock.on.mockImplementation((event: string, handler: any) => { + socketMock.listeners[event] = handler; + }), + })), +})); + +describe("RoomsSocket", () => { + beforeEach(() => { + socketMock.disconnect.mockClear(); + socketMock.emit.mockClear(); + socketMock.on.mockClear(); + socketMock.listeners = {}; + }); + + function createRoomsSocket() { + return RoomsSocket({ + config: { + serverUrl: "https://api.base44.test", + mountPath: "/socket.io/", + transports: ["websocket"], + appId: "test-app-id", + token: "test-token", + }, + }); + } + + test("shares one room join across multiple listeners until the last unsubscribe", () => { + const socket = createRoomsSocket(); + const firstUnsubscribe = socket.subscribeToRoom("room-a", {}); + const secondUnsubscribe = socket.subscribeToRoom("room-a", {}); + + expect(socketMock.emit).toHaveBeenCalledTimes(1); + expect(socketMock.emit).toHaveBeenCalledWith("join", "room-a"); + + firstUnsubscribe(); + + expect(socketMock.emit).toHaveBeenCalledTimes(1); + + secondUnsubscribe(); + + expect(socketMock.emit).toHaveBeenCalledTimes(2); + expect(socketMock.emit).toHaveBeenLastCalledWith("leave", "room-a"); + }); + + test("rejoins a room after its last listener unsubscribes", () => { + const socket = createRoomsSocket(); + const firstUnsubscribe = socket.subscribeToRoom("room-a", {}); + + firstUnsubscribe(); + socket.subscribeToRoom("room-a", {}); + + expect(socketMock.emit).toHaveBeenNthCalledWith(1, "join", "room-a"); + expect(socketMock.emit).toHaveBeenNthCalledWith(2, "leave", "room-a"); + expect(socketMock.emit).toHaveBeenNthCalledWith(3, "join", "room-a"); + }); +}); From 23a5b56f1ae8310939af216d0367c7be2c94120b Mon Sep 17 00:00:00 2001 From: Yoav Farhi Date: Mon, 18 May 2026 11:46:14 +0300 Subject: [PATCH 2/2] fix: make socket room unsubscribe lifecycle idempotent --- src/client.ts | 17 +- src/client.types.ts | 9 +- src/index.ts | 1 - src/modules/entities.ts | 403 +++----------------------- src/modules/entities.types.ts | 37 --- src/utils/socket-utils.ts | 6 + tests/unit/entities-subscribe.test.ts | 273 +---------------- tests/unit/socket-utils.test.ts | 12 + 8 files changed, 67 insertions(+), 691 deletions(-) diff --git a/src/client.ts b/src/client.ts index 8e16bfaa..a028f337 100644 --- a/src/client.ts +++ b/src/client.ts @@ -162,20 +162,11 @@ export function createClient(config: CreateClientConfig): Base44Client { } } - const userAnalyticsModule = createAnalyticsModule({ - axiosClient, - serverUrl, - appId, - userAuthModule, - }); - const userModules = { entities: createEntitiesModule({ axios: axiosClient, appId, getSocket, - subscriptionOptions: options?.entitySubscriptions, - trackSubscriptionEvent: userAnalyticsModule.track, }), integrations: createIntegrationsModule(axiosClient, appId), connectors: createUserConnectorsModule(axiosClient, appId), @@ -201,7 +192,12 @@ export function createClient(config: CreateClientConfig): Base44Client { }), appLogs: createAppLogsModule(axiosClient, appId), users: createUsersModule(axiosClient, appId), - analytics: userAnalyticsModule, + analytics: createAnalyticsModule({ + axiosClient, + serverUrl, + appId, + userAuthModule, + }), cleanup: () => { userModules.analytics.cleanup(); if (socket) { @@ -215,7 +211,6 @@ export function createClient(config: CreateClientConfig): Base44Client { axios: serviceRoleAxiosClient, appId, getSocket, - subscriptionOptions: options?.entitySubscriptions, }), integrations: createIntegrationsModule(serviceRoleAxiosClient, appId), sso: createSsoModule(serviceRoleAxiosClient, appId), diff --git a/src/client.types.ts b/src/client.types.ts index 6f54f413..6b4c9c57 100644 --- a/src/client.types.ts +++ b/src/client.types.ts @@ -1,7 +1,4 @@ -import type { - EntitiesModule, - EntitySubscriptionOptions, -} from "./modules/entities.types.js"; +import type { EntitiesModule } from "./modules/entities.types.js"; import type { IntegrationsModule } from "./modules/integrations.types.js"; import type { AuthModule } from "./modules/auth.types.js"; import type { SsoModule } from "./modules/sso.types.js"; @@ -22,10 +19,6 @@ export interface CreateClientOptions { * Optional error handler that will be called whenever an API error occurs. */ onError?: (error: Error) => void; - /** - * Client-side controls for realtime entity subscriptions. - */ - entitySubscriptions?: EntitySubscriptionOptions; } /** diff --git a/src/index.ts b/src/index.ts index f1d29d4b..bc531d89 100644 --- a/src/index.ts +++ b/src/index.ts @@ -42,7 +42,6 @@ export type { EntityFilterValue, EntityHandler, EntityRecord, - EntitySubscriptionOptions, EntityTypeRegistry, ImportResult, RealtimeEventType, diff --git a/src/modules/entities.ts b/src/modules/entities.ts index dc1523c5..1eaaf287 100644 --- a/src/modules/entities.ts +++ b/src/modules/entities.ts @@ -5,7 +5,6 @@ import { EntitiesModule, EntityFilterQuery, EntityHandler, - EntitySubscriptionOptions, ImportResult, RealtimeCallback, RealtimeEvent, @@ -13,7 +12,6 @@ import { SortField, UpdateManyResult, } from "./entities.types"; -import type { TrackEventParams } from "./analytics.types"; import { RoomsSocket } from "../utils/socket-utils.js"; /** @@ -24,41 +22,6 @@ export interface EntitiesModuleConfig { axios: AxiosInstance; appId: string; getSocket: () => ReturnType; - subscriptionOptions?: EntitySubscriptionOptions; - trackSubscriptionEvent?: (params: TrackEventParams) => void; -} - -const DEFAULT_MAX_ACTIVE_ENTITY_SUBSCRIPTIONS = 100; -const DEFAULT_SUBSCRIPTION_CHURN_WARNING_THRESHOLD = 20; -const DEFAULT_SUBSCRIPTION_CHURN_WINDOW_MS = 60_000; -const DEFAULT_EMPTY_ROOM_GRACE_MS = 1_000; -const ENTITY_SUBSCRIPTION_WARNING_EVENT_NAME = - "__entity_subscription_warning__"; - -type SubscriptionChurnAction = "subscribe" | "unsubscribe"; - -type NormalizedEntitySubscriptionOptions = Required; - -interface EntitySubscriptionManager { - subscribe(entityName: string, callback: RealtimeCallback): () => void; -} - -interface EntitySubscriptionState { - room: string; - entityName: string; - callbacks: Map>; - unsubscribeFromRoom: () => void; - closeTimer: ReturnType | null; -} - -interface SubscriptionChurnRecord { - action: SubscriptionChurnAction; - timestamp: number; -} - -interface SubscriptionChurnState { - events: SubscriptionChurnRecord[]; - lastWarningAt: number | null; } /** @@ -72,14 +35,6 @@ export function createEntitiesModule( config: EntitiesModuleConfig ): EntitiesModule { const { axios, appId, getSocket } = config; - const entityHandlers = new Map>(); - const subscriptionManager = createEntitySubscriptionManager({ - appId, - getSocket, - options: config.subscriptionOptions, - trackSubscriptionEvent: config.trackSubscriptionEvent, - }); - // Using Proxy to dynamically handle entity names return new Proxy( {}, @@ -94,20 +49,8 @@ export function createEntitiesModule( return undefined; } - const cachedHandler = entityHandlers.get(entityName); - if (cachedHandler) { - return cachedHandler; - } - // Create entity handler - const handler = createEntityHandler( - axios, - appId, - entityName, - subscriptionManager - ); - entityHandlers.set(entityName, handler); - return handler; + return createEntityHandler(axios, appId, entityName, getSocket); }, } ) as EntitiesModule; @@ -132,316 +75,13 @@ function parseRealtimeMessage(dataStr: string): RealtimeEvent | null } } -function normalizeEntitySubscriptionOptions( - options?: EntitySubscriptionOptions -): NormalizedEntitySubscriptionOptions { - return { - maxActiveSubscriptions: normalizePositiveInteger( - options?.maxActiveSubscriptions, - DEFAULT_MAX_ACTIVE_ENTITY_SUBSCRIPTIONS - ), - churnWarningThreshold: normalizePositiveInteger( - options?.churnWarningThreshold, - DEFAULT_SUBSCRIPTION_CHURN_WARNING_THRESHOLD - ), - churnWindowMs: normalizePositiveInteger( - options?.churnWindowMs, - DEFAULT_SUBSCRIPTION_CHURN_WINDOW_MS - ), - emptyRoomGraceMs: normalizeNonNegativeInteger( - options?.emptyRoomGraceMs, - DEFAULT_EMPTY_ROOM_GRACE_MS - ), - }; -} - -function normalizePositiveInteger( - value: number | undefined, - fallback: number -): number { - if (typeof value !== "number" || !Number.isFinite(value) || value <= 0) { - return fallback; - } - return Math.floor(value); -} - -function normalizeNonNegativeInteger( - value: number | undefined, - fallback: number -): number { - if (typeof value !== "number" || !Number.isFinite(value) || value < 0) { - return fallback; - } - return Math.floor(value); -} - -function createEntitySubscriptionManager({ - appId, - getSocket, - options, - trackSubscriptionEvent, -}: { - appId: string; - getSocket: () => ReturnType; - options?: EntitySubscriptionOptions; - trackSubscriptionEvent?: (params: TrackEventParams) => void; -}): EntitySubscriptionManager { - const normalizedOptions = normalizeEntitySubscriptionOptions(options); - const activeSubscriptions = new Map(); - const churnByRoom = new Map(); - const roomsWarnedForCap = new Set(); - let nextCallbackId = 1; - - function makeRoom(entityName: string) { - return `entities:${appId}:${entityName}`; - } - - function emitWarning( - message: string, - properties: NonNullable - ) { - console.warn(message); - - try { - trackSubscriptionEvent?.({ - eventName: ENTITY_SUBSCRIPTION_WARNING_EVENT_NAME, - properties, - }); - } catch { - // Diagnostics should never break application code. - } - } - - function recordSubscriptionActivity( - room: string, - entityName: string, - action: SubscriptionChurnAction - ) { - const now = Date.now(); - const churnState = - churnByRoom.get(room) ?? - ({ - events: [], - lastWarningAt: null, - } satisfies SubscriptionChurnState); - - churnState.events = churnState.events.filter( - (event) => now - event.timestamp <= normalizedOptions.churnWindowMs - ); - churnState.events.push({ action, timestamp: now }); - churnByRoom.set(room, churnState); - - const subscribeCount = churnState.events.filter( - (event) => event.action === "subscribe" - ).length; - const unsubscribeCount = churnState.events.length - subscribeCount; - - if ( - churnState.events.length < normalizedOptions.churnWarningThreshold || - subscribeCount === 0 || - unsubscribeCount === 0 || - (churnState.lastWarningAt !== null && - now - churnState.lastWarningAt < normalizedOptions.churnWindowMs) - ) { - return; - } - - churnState.lastWarningAt = now; - emitWarning( - `[Base44 SDK] entities.${entityName}.subscribe() is being created and cleaned up repeatedly ` + - `(${churnState.events.length} subscribe/unsubscribe operations in ` + - `${normalizedOptions.churnWindowMs}ms). Keep realtime subscriptions in a stable lifecycle to avoid socket churn.`, - { - reason: "subscription_churn", - app_id: appId, - entity: entityName, - room, - action, - activity_count: churnState.events.length, - subscribe_count: subscribeCount, - unsubscribe_count: unsubscribeCount, - churn_window_ms: normalizedOptions.churnWindowMs, - empty_room_grace_ms: normalizedOptions.emptyRoomGraceMs, - churn_warning_threshold: - normalizedOptions.churnWarningThreshold, - } - ); - } - - function warnForSubscriptionCap(room: string, entityName: string) { - if (roomsWarnedForCap.has(room)) { - return; - } - - roomsWarnedForCap.add(room); - emitWarning( - `[Base44 SDK] Realtime entity subscription cap reached ` + - `(${normalizedOptions.maxActiveSubscriptions} active entities per SDK client/tab). ` + - `Skipping entities.${entityName}.subscribe(). Unsubscribe from unused entity subscriptions before subscribing to more entities.`, - { - reason: "active_subscription_cap", - app_id: appId, - entity: entityName, - room, - active_subscription_count: activeSubscriptions.size, - max_active_subscriptions: - normalizedOptions.maxActiveSubscriptions, - } - ); - } - - function closeRoomSubscription(state: EntitySubscriptionState) { - clearPendingClose(state); - state.unsubscribeFromRoom(); - activeSubscriptions.delete(state.room); - - if ( - activeSubscriptions.size < normalizedOptions.maxActiveSubscriptions - ) { - roomsWarnedForCap.clear(); - } - } - - function clearPendingClose(state: EntitySubscriptionState) { - if (!state.closeTimer) { - return; - } - - clearTimeout(state.closeTimer); - state.closeTimer = null; - } - - function scheduleRoomClose(state: EntitySubscriptionState) { - if (state.closeTimer) { - return; - } - - if (normalizedOptions.emptyRoomGraceMs === 0) { - closeRoomSubscription(state); - return; - } - - const closeTimer = setTimeout(() => { - state.closeTimer = null; - - if ( - state.callbacks.size === 0 && - activeSubscriptions.get(state.room) === state - ) { - closeRoomSubscription(state); - } - }, normalizedOptions.emptyRoomGraceMs); - - closeTimer.unref?.(); - state.closeTimer = closeTimer; - } - - function dispatchRealtimeMessage( - state: EntitySubscriptionState, - dataStr: string - ) { - if (state.callbacks.size === 0) { - return; - } - - const event = parseRealtimeMessage(dataStr); - if (!event) { - return; - } - - // Server signals oversize broadcasts with `_oversize: true` on - // `data`. The wire payload was slimmed to fit under the realtime - // transport cap, so big string fields arrive as empty strings (or - // the whole record collapses to a stub). Surface this to the - // developer console so they know to fetch the full record on - // demand (e.g. a follow-up entities.X.get(id) call) instead of - // rendering the slimmed payload directly. Skip on delete events - // — the record no longer exists. - if (event.type !== "delete" && (event.data as any)?._oversize) { - console.error( - `[Base44 SDK] Realtime broadcast for ${state.entityName}#${event.id} was oversize and got slimmed for transport. ` + - `Fields >10 KB are empty and the rest of the record may be a stub. ` + - `Call \`entities.${state.entityName}.get("${event.id}")\` to fetch the full record.` - ); - } - - Array.from(state.callbacks.values()).forEach((callback) => { - try { - callback(event); - } catch (error) { - console.error("[Base44 SDK] Subscription callback error:", error); - } - }); - } - - function openRoomSubscription(entityName: string, room: string) { - const state: EntitySubscriptionState = { - room, - entityName, - callbacks: new Map(), - unsubscribeFromRoom: () => {}, - closeTimer: null, - }; - const socket = getSocket(); - - state.unsubscribeFromRoom = socket.subscribeToRoom(room, { - update_model: (msg) => { - dispatchRealtimeMessage(state, msg.data); - }, - }); - activeSubscriptions.set(room, state); - return state; - } - - return { - subscribe(entityName: string, callback: RealtimeCallback) { - const room = makeRoom(entityName); - recordSubscriptionActivity(room, entityName, "subscribe"); - - let state = activeSubscriptions.get(room); - if (!state) { - if ( - activeSubscriptions.size >= - normalizedOptions.maxActiveSubscriptions - ) { - warnForSubscriptionCap(room, entityName); - return () => {}; - } - state = openRoomSubscription(entityName, room); - } else { - clearPendingClose(state); - } - - const callbackId = nextCallbackId++; - state.callbacks.set(callbackId, callback as RealtimeCallback); - - let unsubscribed = false; - return () => { - if (unsubscribed) { - return; - } - unsubscribed = true; - recordSubscriptionActivity(room, entityName, "unsubscribe"); - state.callbacks.delete(callbackId); - - if ( - state.callbacks.size === 0 && - activeSubscriptions.get(room) === state - ) { - scheduleRoomClose(state); - } - }; - }, - }; -} - /** * Creates a handler for a specific entity. * * @param axios - Axios instance * @param appId - Application ID * @param entityName - Entity name - * @param subscriptionManager - Shared realtime subscription manager + * @param getSocket - Function to get the socket instance * @returns Entity handler with CRUD methods * @internal */ @@ -449,7 +89,7 @@ function createEntityHandler( axios: AxiosInstance, appId: string, entityName: string, - subscriptionManager: EntitySubscriptionManager + getSocket: () => ReturnType ): EntityHandler { const baseURL = `/apps/${appId}/entities/${entityName}`; @@ -546,7 +186,42 @@ function createEntityHandler( // Subscribe to realtime updates subscribe(callback: RealtimeCallback): () => void { - return subscriptionManager.subscribe(entityName, callback); + const room = `entities:${appId}:${entityName}`; + + // Get the socket and subscribe to the room + const socket = getSocket(); + const unsubscribe = socket.subscribeToRoom(room, { + update_model: (msg) => { + const event = parseRealtimeMessage(msg.data); + if (!event) { + return; + } + + // Server signals oversize broadcasts with `_oversize: true` on + // `data`. The wire payload was slimmed to fit under the realtime + // transport cap, so big string fields arrive as empty strings (or + // the whole record collapses to a stub). Surface this to the + // developer console so they know to fetch the full record on + // demand (e.g. a follow-up entities.X.get(id) call) instead of + // rendering the slimmed payload directly. Skip on delete events + // — the record no longer exists. + if (event.type !== "delete" && (event.data as any)?._oversize) { + console.error( + `[Base44 SDK] Realtime broadcast for ${entityName}#${event.id} was oversize and got slimmed for transport. ` + + `Fields >10 KB are empty and the rest of the record may be a stub. ` + + `Call \`entities.${entityName}.get("${event.id}")\` to fetch the full record.` + ); + } + + try { + callback(event); + } catch (error) { + console.error("[Base44 SDK] Subscription callback error:", error); + } + }, + }); + + return unsubscribe; }, }; } diff --git a/src/modules/entities.types.ts b/src/modules/entities.types.ts index bab7a86a..c5854dd3 100644 --- a/src/modules/entities.types.ts +++ b/src/modules/entities.types.ts @@ -26,43 +26,6 @@ export interface RealtimeEvent { */ export type RealtimeCallback = (event: RealtimeEvent) => void; -/** - * Client-side controls for realtime entity subscriptions. - */ -export interface EntitySubscriptionOptions { - /** - * Maximum number of distinct active entity realtime subscriptions allowed - * for one SDK client instance. Repeated subscriptions to the same entity - * share one active entity subscription and do not count again. - * - * @defaultValue `100` - */ - maxActiveSubscriptions?: number; - /** - * Number of subscribe/unsubscribe operations for the same entity within - * `churnWindowMs` before the SDK logs a warning and emits diagnostic - * telemetry. - * - * @defaultValue `20` - */ - churnWarningThreshold?: number; - /** - * Time window, in milliseconds, used to detect repeated subscribe/unsubscribe - * churn for one entity. - * - * @defaultValue `60000` - */ - churnWindowMs?: number; - /** - * Grace period, in milliseconds, before the SDK leaves an entity realtime - * room after its last local callback unsubscribes. A new subscription to - * the same entity during this window reuses the existing room membership. - * - * @defaultValue `1000` - */ - emptyRoomGraceMs?: number; -} - /** * Result returned when deleting a single entity. */ diff --git a/src/utils/socket-utils.ts b/src/utils/socket-utils.ts index 14598b15..7a657510 100644 --- a/src/utils/socket-utils.ts +++ b/src/utils/socket-utils.ts @@ -151,8 +151,14 @@ export function RoomsSocket({ config }: { config: RoomsSocketConfig }) { } roomsToListeners[room].push(handlers); + let unsubscribed = false; return () => { + if (unsubscribed) { + return; + } + + unsubscribed = true; roomsToListeners[room] = roomsToListeners[room]?.filter((listener) => listener !== handlers) ?? []; diff --git a/tests/unit/entities-subscribe.test.ts b/tests/unit/entities-subscribe.test.ts index 482cd959..7ac3d905 100644 --- a/tests/unit/entities-subscribe.test.ts +++ b/tests/unit/entities-subscribe.test.ts @@ -7,22 +7,19 @@ describe("Entities Module - subscribe()", () => { // Helper to create a mock socket function createMockSocket() { const listeners: Record = {}; - const unsubscribes: Record> = {}; return { subscribeToRoom: vi.fn((room: string, handlers: any) => { listeners[room] = handlers; - const unsubscribe = vi.fn(() => { + // Return unsubscribe function + return () => { delete listeners[room]; - }); - unsubscribes[room] = unsubscribe; - return unsubscribe; + }; }), // Helper to simulate incoming messages _simulateMessage: (room: string, msg: any) => { listeners[room]?.update_model?.(msg); }, _getListeners: () => listeners, - _getUnsubscribe: (room: string) => unsubscribes[room], }; } @@ -180,270 +177,6 @@ describe("Entities Module - subscribe()", () => { expect(callback).toHaveBeenCalledTimes(1); }); - test("subscribe() should fan out callbacks through one socket room subscription", async () => { - vi.useFakeTimers(); - const mockSocket = createMockSocket(); - const mockAxios = createMockAxios(); - - try { - const entities = createEntitiesModule({ - axios: mockAxios as any, - appId, - getSocket: () => mockSocket as any, - }); - - const firstCallback = vi.fn(); - const secondCallback = vi.fn(); - const firstUnsubscribe = entities.Todo.subscribe(firstCallback); - const secondUnsubscribe = entities.Todo.subscribe(secondCallback); - const room = `entities:${appId}:Todo`; - const roomUnsubscribe = mockSocket._getUnsubscribe(room); - - expect(mockSocket.subscribeToRoom).toHaveBeenCalledTimes(1); - - mockSocket._simulateMessage(room, { - room, - data: JSON.stringify({ - type: "create", - data: { id: "1" }, - id: "1", - timestamp: "2024-01-01T00:00:00.000Z", - }), - }); - - expect(firstCallback).toHaveBeenCalledTimes(1); - expect(secondCallback).toHaveBeenCalledTimes(1); - - firstUnsubscribe(); - - mockSocket._simulateMessage(room, { - room, - data: JSON.stringify({ - type: "update", - data: { id: "1" }, - id: "1", - timestamp: "2024-01-01T00:00:00.000Z", - }), - }); - - expect(firstCallback).toHaveBeenCalledTimes(1); - expect(secondCallback).toHaveBeenCalledTimes(2); - expect(roomUnsubscribe).not.toHaveBeenCalled(); - - secondUnsubscribe(); - - expect(roomUnsubscribe).not.toHaveBeenCalled(); - - await vi.advanceTimersByTimeAsync(1_000); - - expect(roomUnsubscribe).toHaveBeenCalledTimes(1); - } finally { - vi.useRealTimers(); - } - }); - - test("subscribe() should cancel the empty-room leave when resubscribed during grace", async () => { - vi.useFakeTimers(); - const mockSocket = createMockSocket(); - const mockAxios = createMockAxios(); - - try { - const entities = createEntitiesModule({ - axios: mockAxios as any, - appId, - getSocket: () => mockSocket as any, - subscriptionOptions: { emptyRoomGraceMs: 1_000 }, - }); - const room = `entities:${appId}:Todo`; - const firstUnsubscribe = entities.Todo.subscribe(vi.fn()); - const roomUnsubscribe = mockSocket._getUnsubscribe(room); - - firstUnsubscribe(); - - expect(roomUnsubscribe).not.toHaveBeenCalled(); - - const secondCallback = vi.fn(); - const secondUnsubscribe = entities.Todo.subscribe(secondCallback); - - await vi.advanceTimersByTimeAsync(1_000); - - expect(mockSocket.subscribeToRoom).toHaveBeenCalledTimes(1); - expect(roomUnsubscribe).not.toHaveBeenCalled(); - - mockSocket._simulateMessage(room, { - room, - data: JSON.stringify({ - type: "update", - data: { id: "1" }, - id: "1", - timestamp: "2024-01-01T00:00:00.000Z", - }), - }); - - expect(secondCallback).toHaveBeenCalledTimes(1); - - secondUnsubscribe(); - await vi.advanceTimersByTimeAsync(1_000); - - expect(roomUnsubscribe).toHaveBeenCalledTimes(1); - } finally { - vi.useRealTimers(); - } - }); - - test("subscribe() should leave an empty room after the grace period expires", async () => { - vi.useFakeTimers(); - const mockSocket = createMockSocket(); - const mockAxios = createMockAxios(); - - try { - const entities = createEntitiesModule({ - axios: mockAxios as any, - appId, - getSocket: () => mockSocket as any, - subscriptionOptions: { emptyRoomGraceMs: 1_000 }, - }); - const room = `entities:${appId}:Todo`; - const unsubscribe = entities.Todo.subscribe(vi.fn()); - const roomUnsubscribe = mockSocket._getUnsubscribe(room); - - unsubscribe(); - await vi.advanceTimersByTimeAsync(999); - - expect(roomUnsubscribe).not.toHaveBeenCalled(); - - await vi.advanceTimersByTimeAsync(1); - - expect(roomUnsubscribe).toHaveBeenCalledTimes(1); - } finally { - vi.useRealTimers(); - } - }); - - test("subscribe() should cap distinct active entity subscriptions", () => { - const mockSocket = createMockSocket(); - const mockAxios = createMockAxios(); - const trackSubscriptionEvent = vi.fn(); - const warnSpy = vi.spyOn(console, "warn").mockImplementation(() => {}); - - const entities = createEntitiesModule({ - axios: mockAxios as any, - appId, - getSocket: () => mockSocket as any, - subscriptionOptions: { maxActiveSubscriptions: 1 }, - trackSubscriptionEvent, - }); - - const todoCallback = vi.fn(); - const userCallback = vi.fn(); - const unsubscribeTodo = entities.Todo.subscribe(todoCallback); - const unsubscribeUser = entities.User.subscribe(userCallback); - - expect(mockSocket.subscribeToRoom).toHaveBeenCalledTimes(1); - expect(mockSocket.subscribeToRoom).toHaveBeenCalledWith( - `entities:${appId}:Todo`, - expect.any(Object) - ); - expect(warnSpy).toHaveBeenCalledWith( - expect.stringContaining("Realtime entity subscription cap reached") - ); - expect(trackSubscriptionEvent).toHaveBeenCalledWith({ - eventName: "__entity_subscription_warning__", - properties: expect.objectContaining({ - reason: "active_subscription_cap", - entity: "User", - active_subscription_count: 1, - max_active_subscriptions: 1, - }), - }); - - mockSocket._simulateMessage(`entities:${appId}:User`, { - room: `entities:${appId}:User`, - data: JSON.stringify({ - type: "create", - data: { id: "blocked" }, - id: "blocked", - timestamp: "2024-01-01T00:00:00.000Z", - }), - }); - - expect(userCallback).not.toHaveBeenCalled(); - - unsubscribeUser(); - unsubscribeTodo(); - warnSpy.mockRestore(); - }); - - test("subscribe() should warn and emit telemetry on repeated subscription churn", () => { - const mockSocket = createMockSocket(); - const mockAxios = createMockAxios(); - const trackSubscriptionEvent = vi.fn(); - const warnSpy = vi.spyOn(console, "warn").mockImplementation(() => {}); - - const entities = createEntitiesModule({ - axios: mockAxios as any, - appId, - getSocket: () => mockSocket as any, - subscriptionOptions: { - churnWarningThreshold: 4, - churnWindowMs: 60_000, - }, - trackSubscriptionEvent, - }); - - const firstUnsubscribe = entities.Todo.subscribe(vi.fn()); - firstUnsubscribe(); - const secondUnsubscribe = entities.Todo.subscribe(vi.fn()); - secondUnsubscribe(); - - expect(warnSpy).toHaveBeenCalledWith( - expect.stringContaining("created and cleaned up repeatedly") - ); - expect(trackSubscriptionEvent).toHaveBeenCalledWith({ - eventName: "__entity_subscription_warning__", - properties: expect.objectContaining({ - reason: "subscription_churn", - entity: "Todo", - activity_count: 4, - subscribe_count: 2, - unsubscribe_count: 2, - churn_window_ms: 60000, - churn_warning_threshold: 4, - }), - }); - - warnSpy.mockRestore(); - }); - - test("subscribe() should not warn when many callbacks fan out without unsubscribe churn", () => { - const mockSocket = createMockSocket(); - const mockAxios = createMockAxios(); - const trackSubscriptionEvent = vi.fn(); - const warnSpy = vi.spyOn(console, "warn").mockImplementation(() => {}); - - const entities = createEntitiesModule({ - axios: mockAxios as any, - appId, - getSocket: () => mockSocket as any, - subscriptionOptions: { - churnWarningThreshold: 4, - churnWindowMs: 60_000, - }, - trackSubscriptionEvent, - }); - - entities.Todo.subscribe(vi.fn()); - entities.Todo.subscribe(vi.fn()); - entities.Todo.subscribe(vi.fn()); - entities.Todo.subscribe(vi.fn()); - - expect(mockSocket.subscribeToRoom).toHaveBeenCalledTimes(1); - expect(warnSpy).not.toHaveBeenCalled(); - expect(trackSubscriptionEvent).not.toHaveBeenCalled(); - - warnSpy.mockRestore(); - }); - test("subscribe() should not call callback for invalid JSON messages", () => { const mockSocket = createMockSocket(); const mockAxios = createMockAxios(); diff --git a/tests/unit/socket-utils.test.ts b/tests/unit/socket-utils.test.ts index 116183b6..035f3789 100644 --- a/tests/unit/socket-utils.test.ts +++ b/tests/unit/socket-utils.test.ts @@ -68,4 +68,16 @@ describe("RoomsSocket", () => { expect(socketMock.emit).toHaveBeenNthCalledWith(2, "leave", "room-a"); expect(socketMock.emit).toHaveBeenNthCalledWith(3, "join", "room-a"); }); + + test("unsubscribe is idempotent after the room is left", () => { + const socket = createRoomsSocket(); + const unsubscribe = socket.subscribeToRoom("room-a", {}); + + unsubscribe(); + unsubscribe(); + + expect(socketMock.emit).toHaveBeenCalledTimes(2); + expect(socketMock.emit).toHaveBeenNthCalledWith(1, "join", "room-a"); + expect(socketMock.emit).toHaveBeenNthCalledWith(2, "leave", "room-a"); + }); });