diff --git a/packages/twenty-server/src/engine/core-modules/cache-storage/services/__tests__/cache-storage.service.spec.ts b/packages/twenty-server/src/engine/core-modules/cache-storage/services/__tests__/cache-storage.service.spec.ts index fc68a09cf6..27ef73bfa2 100644 --- a/packages/twenty-server/src/engine/core-modules/cache-storage/services/__tests__/cache-storage.service.spec.ts +++ b/packages/twenty-server/src/engine/core-modules/cache-storage/services/__tests__/cache-storage.service.spec.ts @@ -109,4 +109,90 @@ describe('CacheStorageService', () => { ); }); }); + + describe('sorted sets', () => { + const createRedisCacheMock = () => { + const zAdd = jest.fn().mockResolvedValue(1); + const zRem = jest.fn().mockResolvedValue(1); + const exec = jest.fn().mockResolvedValue([2, 3]); + const multi = { + zRemRangeByScore: jest.fn(), + zCard: jest.fn(), + exec, + }; + + multi.zRemRangeByScore.mockReturnValue(multi); + multi.zCard.mockReturnValue(multi); + + const cache = { + store: { + name: 'redis', + client: { + zAdd, + zRem, + multi: jest.fn().mockReturnValue(multi), + }, + }, + } as unknown as Cache; + + return { cache, exec, multi, zAdd, zRem }; + }; + + it('adds members with scores', async () => { + const { cache, zAdd } = createRedisCacheMock(); + const cacheStorageService = new CacheStorageService( + cache, + CacheStorageNamespace.EngineWorkspace, + ); + const entries = [ + { score: 1000, value: 'stream-1' }, + { score: 2000, value: 'stream-2' }, + ]; + + await cacheStorageService.sortedSetAdd('active-streams', entries); + + expect(zAdd).toHaveBeenCalledWith(prefixKey('active-streams'), entries); + }); + + it('removes members', async () => { + const { cache, zRem } = createRedisCacheMock(); + const cacheStorageService = new CacheStorageService( + cache, + CacheStorageNamespace.EngineWorkspace, + ); + + await cacheStorageService.sortedSetRemove('active-streams', [ + 'stream-1', + 'stream-2', + ]); + + expect(zRem).toHaveBeenCalledWith(prefixKey('active-streams'), [ + 'stream-1', + 'stream-2', + ]); + }); + + it('removes expired members and returns the remaining count atomically', async () => { + const { cache, exec, multi } = createRedisCacheMock(); + const cacheStorageService = new CacheStorageService( + cache, + CacheStorageNamespace.EngineWorkspace, + ); + + const count = await cacheStorageService.sortedSetRemoveByScoreAndCount( + 'active-streams', + 0, + 1000, + ); + + expect(multi.zRemRangeByScore).toHaveBeenCalledWith( + prefixKey('active-streams'), + 0, + 1000, + ); + expect(multi.zCard).toHaveBeenCalledWith(prefixKey('active-streams')); + expect(exec).toHaveBeenCalledTimes(1); + expect(count).toBe(3); + }); + }); }); diff --git a/packages/twenty-server/src/engine/core-modules/cache-storage/services/cache-storage.service.ts b/packages/twenty-server/src/engine/core-modules/cache-storage/services/cache-storage.service.ts index 0068776c94..b29f8c08ef 100644 --- a/packages/twenty-server/src/engine/core-modules/cache-storage/services/cache-storage.service.ts +++ b/packages/twenty-server/src/engine/core-modules/cache-storage/services/cache-storage.service.ts @@ -236,45 +236,56 @@ export class CacheStorageService { } while (cursor !== 0); } - async scanAndCountSetMembers(scanPattern: string): Promise { + async sortedSetAdd( + key: string, + entries: Array<{ score: number; value: string }>, + ): Promise { + if (entries.length === 0) { + return 0; + } + + if (!this.isRedisCache(this.cache)) { + throw new Error('sortedSetAdd is only supported with Redis cache'); + } + + return this.cache.store.client.zAdd(this.getKey(key), entries); + } + + async sortedSetRemove(key: string, values: string[]): Promise { + if (values.length === 0) { + return 0; + } + + if (!this.isRedisCache(this.cache)) { + throw new Error('sortedSetRemove is only supported with Redis cache'); + } + + return this.cache.store.client.zRem(this.getKey(key), values); + } + + async sortedSetRemoveByScoreAndCount( + key: string, + minScore: number, + maxScore: number, + ): Promise { if (!this.isRedisCache(this.cache)) { throw new Error( - 'scanAndCountSetMembers is only supported with Redis cache', + 'sortedSetRemoveByScoreAndCount is only supported with Redis cache', ); } - const redisClient = this.cache.store.client; - let cursor = 0; - let totalCount = 0; + const prefixedKey = this.getKey(key); + const [, count] = await this.cache.store.client + .multi() + .zRemRangeByScore(prefixedKey, minScore, maxScore) + .zCard(prefixedKey) + .exec(); - do { - const result = await redisClient.scan(cursor, { - MATCH: `${this.namespace}:${scanPattern}`, - COUNT: 100, - }); + if (count instanceof Error) { + throw count; + } - cursor = result.cursor; - const keys = result.keys; - - if (keys.length > 0) { - const pipeline = redisClient.multi(); - - for (const key of keys) { - pipeline.sCard(key); - } - - const results = await pipeline.exec(); - - for (const result of results) { - if (result instanceof Error) { - throw result; - } - totalCount += result as number; - } - } - } while (cursor !== 0); - - return totalCount; + return count as number; } async acquireLock(key: string, ttl = 1000): Promise { diff --git a/packages/twenty-server/src/engine/subscriptions/__tests__/event-stream.service.spec.ts b/packages/twenty-server/src/engine/subscriptions/__tests__/event-stream.service.spec.ts new file mode 100644 index 0000000000..140fd1619e --- /dev/null +++ b/packages/twenty-server/src/engine/subscriptions/__tests__/event-stream.service.spec.ts @@ -0,0 +1,187 @@ +import { type CacheStorageService } from 'src/engine/core-modules/cache-storage/services/cache-storage.service'; +import { type MetricsService } from 'src/engine/core-modules/metrics/metrics.service'; +import { EVENT_STREAM_TTL_MS } from 'src/engine/subscriptions/constants/event-stream-ttl.constant'; +import { EventStreamService } from 'src/engine/subscriptions/event-stream.service'; + +const WORKSPACE_ID = 'workspace-id'; +const EVENT_STREAM_CHANNEL_ID = 'event-stream-channel-id'; +const EVENT_STREAM_KEY = `eventStream:${WORKSPACE_ID}:${EVENT_STREAM_CHANNEL_ID}`; +const ACTIVE_STREAMS_KEY = `workspace:${WORKSPACE_ID}:activeStreams`; +const ACTIVE_STREAM_EXPIRATIONS_KEY = 'activeStreamExpirations'; +const ACTIVE_STREAM_EXPIRATION_MEMBER = `${WORKSPACE_ID}:${EVENT_STREAM_CHANNEL_ID}`; + +describe('EventStreamService', () => { + let cacheStorageService: jest.Mocked; + let service: EventStreamService; + + beforeEach(() => { + jest.useFakeTimers(); + jest.setSystemTime(new Date('2026-07-26T12:00:00.000Z')); + + cacheStorageService = { + del: jest.fn(), + expire: jest.fn(), + get: jest.fn(), + set: jest.fn(), + setAdd: jest.fn(), + setRemove: jest.fn(), + sortedSetAdd: jest.fn(), + sortedSetRemove: jest.fn(), + sortedSetRemoveByScoreAndCount: jest.fn(), + } as unknown as jest.Mocked; + + service = new EventStreamService(cacheStorageService, { + createObservableGauge: jest.fn(), + } as unknown as MetricsService); + }); + + afterEach(() => { + jest.useRealTimers(); + }); + + it('counts live streams without scanning Redis keys', async () => { + cacheStorageService.sortedSetRemoveByScoreAndCount.mockResolvedValue(3); + + await expect(service.getTotalActiveStreamCount()).resolves.toBe(3); + await expect(service.getTotalActiveStreamCount()).resolves.toBe(3); + + expect( + cacheStorageService.sortedSetRemoveByScoreAndCount, + ).toHaveBeenCalledTimes(1); + expect( + cacheStorageService.sortedSetRemoveByScoreAndCount, + ).toHaveBeenCalledWith(ACTIVE_STREAM_EXPIRATIONS_KEY, 0, Date.now()); + }); + + it('adds a stream to its functional and metric indexes on creation', async () => { + cacheStorageService.get.mockResolvedValue(undefined); + + await service.createEventStream({ + workspaceId: WORKSPACE_ID, + eventStreamChannelId: EVENT_STREAM_CHANNEL_ID, + authContext: { userId: 'user-id' }, + }); + + expect(cacheStorageService.set).toHaveBeenCalledWith( + EVENT_STREAM_KEY, + { + authContext: { userId: 'user-id' }, + workspaceId: WORKSPACE_ID, + queries: {}, + createdAt: Date.now(), + }, + EVENT_STREAM_TTL_MS, + ); + expect(cacheStorageService.setAdd).toHaveBeenCalledWith( + ACTIVE_STREAMS_KEY, + [EVENT_STREAM_CHANNEL_ID], + EVENT_STREAM_TTL_MS, + ); + expect(cacheStorageService.sortedSetAdd).toHaveBeenCalledWith( + ACTIVE_STREAM_EXPIRATIONS_KEY, + [ + { + score: Date.now() + EVENT_STREAM_TTL_MS, + value: ACTIVE_STREAM_EXPIRATION_MEMBER, + }, + ], + ); + }); + + it('tracks a stream after a successful refresh', async () => { + cacheStorageService.expire + .mockResolvedValueOnce(true) + .mockResolvedValueOnce(true); + + await expect( + service.refreshEventStreamTTL({ + workspaceId: WORKSPACE_ID, + eventStreamChannelId: EVENT_STREAM_CHANNEL_ID, + }), + ).resolves.toBe(true); + + expect(cacheStorageService.sortedSetAdd).toHaveBeenCalledWith( + ACTIVE_STREAM_EXPIRATIONS_KEY, + [ + { + score: Date.now() + EVENT_STREAM_TTL_MS, + value: ACTIVE_STREAM_EXPIRATION_MEMBER, + }, + ], + ); + expect(cacheStorageService.sortedSetRemove).not.toHaveBeenCalled(); + }); + + it('does not track a stream whose key is missing during refresh', async () => { + cacheStorageService.expire + .mockResolvedValueOnce(false) + .mockResolvedValueOnce(true); + + await expect( + service.refreshEventStreamTTL({ + workspaceId: WORKSPACE_ID, + eventStreamChannelId: EVENT_STREAM_CHANNEL_ID, + }), + ).resolves.toBe(false); + + expect(cacheStorageService.sortedSetAdd).not.toHaveBeenCalled(); + expect(cacheStorageService.sortedSetRemove).not.toHaveBeenCalled(); + }); + + it('keeps a valid stream tracked when the workspace stream set is missing', async () => { + cacheStorageService.expire + .mockResolvedValueOnce(true) + .mockResolvedValueOnce(false); + + await expect( + service.refreshEventStreamTTL({ + workspaceId: WORKSPACE_ID, + eventStreamChannelId: EVENT_STREAM_CHANNEL_ID, + }), + ).resolves.toBe(false); + + expect(cacheStorageService.sortedSetAdd).toHaveBeenCalledWith( + ACTIVE_STREAM_EXPIRATIONS_KEY, + [ + { + score: Date.now() + EVENT_STREAM_TTL_MS, + value: ACTIVE_STREAM_EXPIRATION_MEMBER, + }, + ], + ); + expect(cacheStorageService.sortedSetRemove).not.toHaveBeenCalled(); + }); + + it('removes a destroyed stream from its functional and metric indexes', async () => { + await service.destroyEventStream({ + workspaceId: WORKSPACE_ID, + eventStreamChannelId: EVENT_STREAM_CHANNEL_ID, + }); + + expect(cacheStorageService.del).toHaveBeenCalledWith(EVENT_STREAM_KEY); + expect(cacheStorageService.setRemove).toHaveBeenCalledWith( + ACTIVE_STREAMS_KEY, + [EVENT_STREAM_CHANNEL_ID], + ); + expect(cacheStorageService.sortedSetRemove).toHaveBeenCalledWith( + ACTIVE_STREAM_EXPIRATIONS_KEY, + [ACTIVE_STREAM_EXPIRATION_MEMBER], + ); + }); + + it('removes stale streams from the functional and metric indexes', async () => { + await service.removeFromActiveStreams(WORKSPACE_ID, [ + EVENT_STREAM_CHANNEL_ID, + 'stale-stream-id', + ]); + + expect(cacheStorageService.setRemove).toHaveBeenCalledWith( + ACTIVE_STREAMS_KEY, + [EVENT_STREAM_CHANNEL_ID, 'stale-stream-id'], + ); + expect(cacheStorageService.sortedSetRemove).toHaveBeenCalledWith( + ACTIVE_STREAM_EXPIRATIONS_KEY, + [ACTIVE_STREAM_EXPIRATION_MEMBER, `${WORKSPACE_ID}:stale-stream-id`], + ); + }); +}); diff --git a/packages/twenty-server/src/engine/subscriptions/event-stream.service.ts b/packages/twenty-server/src/engine/subscriptions/event-stream.service.ts index d043d255b5..47893bb15a 100644 --- a/packages/twenty-server/src/engine/subscriptions/event-stream.service.ts +++ b/packages/twenty-server/src/engine/subscriptions/event-stream.service.ts @@ -19,6 +19,7 @@ import { } from 'src/engine/subscriptions/types/event-stream-data.type'; const ACTIVE_STREAM_COUNT_REFRESH_MS = 5 * 60 * 1_000; +const ACTIVE_STREAM_EXPIRATIONS_KEY = 'activeStreamExpirations'; @Injectable() export class EventStreamService implements OnModuleInit { @@ -50,8 +51,10 @@ export class EventStreamService implements OnModuleInit { if (isStale) { this.activeStreamCount = - await this.cacheStorageService.scanAndCountSetMembers( - 'workspace:*:activeStreams', + await this.cacheStorageService.sortedSetRemoveByScoreAndCount( + ACTIVE_STREAM_EXPIRATIONS_KEY, + 0, + now, ); this.activeStreamCountRefreshedAt = now; } @@ -93,6 +96,11 @@ export class EventStreamService implements OnModuleInit { [eventStreamChannelId], EVENT_STREAM_TTL_MS, ); + + await this.trackActiveStream({ + workspaceId, + eventStreamChannelId, + }); } async destroyEventStream({ @@ -110,6 +118,8 @@ export class EventStreamService implements OnModuleInit { this.getActiveStreamsKey(workspaceId), [eventStreamChannelId], ); + + await this.untrackActiveStreams(workspaceId, [eventStreamChannelId]); } async getActiveStreamIds(workspaceId: string): Promise { @@ -130,6 +140,8 @@ export class EventStreamService implements OnModuleInit { this.getActiveStreamsKey(workspaceId), streamIdsToRemove, ); + + await this.untrackActiveStreams(workspaceId, streamIdsToRemove); } async getStreamsData( @@ -235,9 +247,67 @@ export class EventStreamService implements OnModuleInit { this.cacheStorageService.expire(activeStreamsKey, EVENT_STREAM_TTL_MS), ]); + if (eventStreamRefreshed) { + await this.trackActiveStream({ + workspaceId, + eventStreamChannelId, + }); + } + return eventStreamRefreshed && activeStreamsRefreshed; } + private async trackActiveStream({ + workspaceId, + eventStreamChannelId, + }: { + workspaceId: string; + eventStreamChannelId: string; + }): Promise { + try { + await this.cacheStorageService.sortedSetAdd( + ACTIVE_STREAM_EXPIRATIONS_KEY, + [ + { + score: Date.now() + EVENT_STREAM_TTL_MS, + value: this.getActiveStreamExpirationMember( + workspaceId, + eventStreamChannelId, + ), + }, + ], + ); + } catch (error) { + this.logger.warn(`Failed to track active event stream: ${error}`); + } + } + + private async untrackActiveStreams( + workspaceId: string, + eventStreamChannelIds: string[], + ): Promise { + try { + await this.cacheStorageService.sortedSetRemove( + ACTIVE_STREAM_EXPIRATIONS_KEY, + eventStreamChannelIds.map((eventStreamChannelId) => + this.getActiveStreamExpirationMember( + workspaceId, + eventStreamChannelId, + ), + ), + ); + } catch (error) { + this.logger.warn(`Failed to untrack active event streams: ${error}`); + } + } + + private getActiveStreamExpirationMember( + workspaceId: string, + eventStreamChannelId: string, + ): string { + return `${workspaceId}:${eventStreamChannelId}`; + } + private getEventStreamKey( workspaceId: string, eventStreamId: string,