From 0a2d42e79f335e9a1c4a460c8fc5e0348eb8e3b5 Mon Sep 17 00:00:00 2001 From: Weiko Date: Fri, 21 Nov 2025 12:44:24 +0100 Subject: [PATCH] Implement workspace cache storage (#15962) ## Context Implementing a single service managing all the cache scoped to a workspace, this will be dynamically injected as a WorkspaceContext in the app during a request lifetime (ingested by the future global datasource for example). Usage: ```typescript this.globalWorkspaceOrmManager.executeInWorkspaceContext( authContext, async () => { // Everything here will have access to a workspaceContext, containing all the cache data + a ready to use datasource with workspace scoped metadata, permissions, feature flags, etc... // Internally will call loadWorkspaceContext } ``` Note: executeInWorkspaceContext will probably be owned by a higher level service later and not only the ORM. ```typescript private async loadWorkspaceContext( authContext: WorkspaceAuthContext, ): Promise { const workspaceId = authContext.workspace.id; const cache = await this.workspaceContextCacheService.get( workspaceId, [ 'objectMetadataMaps', 'metadataVersion', 'featureFlagsMap', 'permissionsPerRoleId', ], ); return { authContext, objectMetadataMaps: cache.objectMetadataMaps, metadataVersion: cache.metadataVersion, featureFlagsMap: cache.featureFlagsMap, permissionsPerRoleId: cache.permissionsPerRoleId, }; } ``` The cache retrieval strategy is as followed: ``` - Check if there is an ongoing promise fetching data from the cache => return the promise. - Check in the local cache entry if lastCheckedAt has expired. If not, return as it is without querying redis. - Check in redis the cache entry hash and compare with local cache entry hash, if they are the same return the local cache entry data - Check in redis the cache entry data, if it's there return it and store it into the local cache entry data and update local cache entry hash. If it's not there recompute the data by querying the DB and update both redis and local cache ``` --- .../services/cache-storage.service.ts | 45 +++ ...orkspace-feature-flags-map-cache.module.ts | 1 - .../global-workspace-datasource.module.ts | 2 +- .../decorators/workspace-cache.decorator.ts | 6 + .../services/workspace-cache.service.ts | 365 ++++++++++++++++++ .../workspace-context-cache-entry.type.ts | 5 + .../types/workspace-context-data.type.ts | 11 + .../workspace-cache-provider.service.ts | 6 + .../workspace-cache/workspace-cache.module.ts | 12 + 9 files changed, 451 insertions(+), 2 deletions(-) create mode 100644 packages/twenty-server/src/engine/workspace-cache/decorators/workspace-cache.decorator.ts create mode 100644 packages/twenty-server/src/engine/workspace-cache/services/workspace-cache.service.ts create mode 100644 packages/twenty-server/src/engine/workspace-cache/types/workspace-context-cache-entry.type.ts create mode 100644 packages/twenty-server/src/engine/workspace-cache/types/workspace-context-data.type.ts create mode 100644 packages/twenty-server/src/engine/workspace-cache/workspace-cache-provider.service.ts create mode 100644 packages/twenty-server/src/engine/workspace-cache/workspace-cache.module.ts 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 ace12db201d..ca4ed172ede 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 @@ -28,6 +28,51 @@ export class CacheStorageService { return this.cache.del(this.getKey(key)); } + async mget(keys: string[]): Promise<(T | undefined)[]> { + if (this.isRedisCache()) { + const prefixedKeys = keys.map((k) => this.getKey(k)); + const values = await (this.cache as RedisCache).store.client.mGet( + prefixedKeys, + ); + + return values.map((v) => { + if (v === null || v === undefined) return undefined; + try { + return JSON.parse(v) as T; + } catch { + return v as T; + } + }); + } + + return Promise.all(keys.map((k) => this.get(k))); + } + + async mset( + entries: Array<{ key: string; value: T; ttl?: Milliseconds }>, + ): Promise { + if (this.isRedisCache()) { + const pipeline = (this.cache as RedisCache).store.client.multi(); + + entries.forEach(({ key, value, ttl }) => { + const prefixedKey = this.getKey(key); + + pipeline.set(prefixedKey, JSON.stringify(value)); + if (ttl) { + pipeline.expire(prefixedKey, Math.floor(ttl / 1000)); + } + }); + + await pipeline.exec(); + + return; + } + + await Promise.all( + entries.map(({ key, value, ttl }) => this.set(key, value, ttl)), + ); + } + async setAdd(key: string, value: string[], ttl?: Milliseconds) { if (value.length === 0) { return; diff --git a/packages/twenty-server/src/engine/metadata-modules/workspace-feature-flags-map-cache/workspace-feature-flags-map-cache.module.ts b/packages/twenty-server/src/engine/metadata-modules/workspace-feature-flags-map-cache/workspace-feature-flags-map-cache.module.ts index d8b51b22c5d..881c37a1d4f 100644 --- a/packages/twenty-server/src/engine/metadata-modules/workspace-feature-flags-map-cache/workspace-feature-flags-map-cache.module.ts +++ b/packages/twenty-server/src/engine/metadata-modules/workspace-feature-flags-map-cache/workspace-feature-flags-map-cache.module.ts @@ -10,7 +10,6 @@ import { WorkspaceCacheStorageModule } from 'src/engine/workspace-cache-storage/ imports: [ TypeOrmModule.forFeature([WorkspaceEntity, FeatureFlagEntity]), WorkspaceCacheStorageModule, - WorkspaceCacheStorageModule, ], providers: [WorkspaceFeatureFlagsMapCacheService], exports: [WorkspaceFeatureFlagsMapCacheService], diff --git a/packages/twenty-server/src/engine/twenty-orm/global-workspace-datasource/global-workspace-datasource.module.ts b/packages/twenty-server/src/engine/twenty-orm/global-workspace-datasource/global-workspace-datasource.module.ts index 09f4f41b4ad..f8917e41969 100644 --- a/packages/twenty-server/src/engine/twenty-orm/global-workspace-datasource/global-workspace-datasource.module.ts +++ b/packages/twenty-server/src/engine/twenty-orm/global-workspace-datasource/global-workspace-datasource.module.ts @@ -11,8 +11,8 @@ import { WorkspacePermissionsCacheModule } from 'src/engine/metadata-modules/wor import { EntitySchemaColumnFactory } from 'src/engine/twenty-orm/factories/entity-schema-column.factory'; import { EntitySchemaRelationFactory } from 'src/engine/twenty-orm/factories/entity-schema-relation.factory'; import { EntitySchemaFactory } from 'src/engine/twenty-orm/factories/entity-schema.factory'; -import { GlobalWorkspaceOrmManager } from 'src/engine/twenty-orm/global-workspace-datasource/global-workspace-orm.manager'; import { GlobalWorkspaceDataSourceService } from 'src/engine/twenty-orm/global-workspace-datasource/global-workspace-datasource.service'; +import { GlobalWorkspaceOrmManager } from 'src/engine/twenty-orm/global-workspace-datasource/global-workspace-orm.manager'; import { WorkspaceCacheStorageModule } from 'src/engine/workspace-cache-storage/workspace-cache-storage.module'; import { WorkspaceEventEmitterModule } from 'src/engine/workspace-event-emitter/workspace-event-emitter.module'; diff --git a/packages/twenty-server/src/engine/workspace-cache/decorators/workspace-cache.decorator.ts b/packages/twenty-server/src/engine/workspace-cache/decorators/workspace-cache.decorator.ts new file mode 100644 index 00000000000..5928bdcac0d --- /dev/null +++ b/packages/twenty-server/src/engine/workspace-cache/decorators/workspace-cache.decorator.ts @@ -0,0 +1,6 @@ +import { SetMetadata } from '@nestjs/common'; + +export const WORKSPACE_CACHE_KEY = 'WORKSPACE_CACHE_KEY'; + +export const WorkspaceCache = (workspaceCacheKey: string) => + SetMetadata(WORKSPACE_CACHE_KEY, workspaceCacheKey); diff --git a/packages/twenty-server/src/engine/workspace-cache/services/workspace-cache.service.ts b/packages/twenty-server/src/engine/workspace-cache/services/workspace-cache.service.ts new file mode 100644 index 00000000000..f70aa7252d0 --- /dev/null +++ b/packages/twenty-server/src/engine/workspace-cache/services/workspace-cache.service.ts @@ -0,0 +1,365 @@ +import { Injectable, OnModuleInit } from '@nestjs/common'; +import { DiscoveryService, Reflector } from '@nestjs/core'; + +import crypto from 'crypto'; + +import { isDefined } from 'twenty-shared/utils'; + +import { InjectCacheStorage } from 'src/engine/core-modules/cache-storage/decorators/cache-storage.decorator'; +import { CacheStorageService } from 'src/engine/core-modules/cache-storage/services/cache-storage.service'; +import { CacheStorageNamespace } from 'src/engine/core-modules/cache-storage/types/cache-storage-namespace.enum'; +import { PromiseMemoizer } from 'src/engine/twenty-orm/storage/promise-memoizer.storage'; +import { WORKSPACE_CACHE_KEY } from 'src/engine/workspace-cache/decorators/workspace-cache.decorator'; +import { WorkspaceContextLocalCacheEntry } from 'src/engine/workspace-cache/types/workspace-context-cache-entry.type'; +import { WorkspaceCacheProvider } from 'src/engine/workspace-cache/workspace-cache-provider.service'; + +const LOCAL_STALENESS_TTL_MS = 30_000; +const MEMOIZER_TTL_MS = 10_000; + +@Injectable() +export class WorkspaceCacheService implements OnModuleInit { + private readonly localCache = new Map< + string, + WorkspaceContextLocalCacheEntry + >(); + private readonly workspaceCacheProviders = new Map< + string, + WorkspaceCacheProvider + >(); + private readonly memoizer = new PromiseMemoizer>( + MEMOIZER_TTL_MS, + ); + + constructor( + @InjectCacheStorage(CacheStorageNamespace.EngineWorkspace) + private readonly cacheStorage: CacheStorageService, + private readonly discoveryService: DiscoveryService, + private readonly reflector: Reflector, + ) {} + + async onModuleInit() { + const providers = this.discoveryService.getProviders(); + + for (const wrapper of providers) { + const { instance } = wrapper; + + if (!isDefined(instance) || typeof instance !== 'object') { + continue; + } + + const workspaceCacheKey = this.reflector.get( + WORKSPACE_CACHE_KEY, + instance.constructor, + ); + + if ( + isDefined(workspaceCacheKey) && + instance instanceof WorkspaceCacheProvider + ) { + this.workspaceCacheProviders.set(workspaceCacheKey, instance); + } + } + } + + async getOrRecompute< + T extends Record = Record, + >(workspaceId: string, workspaceCacheKeys: string[]): Promise { + const memoKey = + `${workspaceId}-${[...workspaceCacheKeys].sort().join(',')}` as const; + + const result = await this.memoizer.memoizePromiseAndExecute( + memoKey, + async () => { + const freshResult: Record = {}; + + const { freshKeys, staleKeys } = this.partitionKeysByTTLStaleness( + workspaceId, + workspaceCacheKeys, + ); + + for (const workspaceCacheKey of freshKeys) { + const localKey = this.getCacheKey(workspaceId, workspaceCacheKey); + const cached = this.localCache.get(localKey); + + freshResult[workspaceCacheKey] = cached?.data; + } + + if (staleKeys.length === 0) { + return freshResult; + } + + const staleResults = await this.resolveStaleKeys( + workspaceId, + staleKeys, + ); + + return { ...freshResult, ...staleResults }; + }, + ); + + return result as T; + } + + invalidate(workspaceId: string, workspaceCacheKeys?: string[]): void { + if (!isDefined(workspaceCacheKeys) || workspaceCacheKeys.length === 0) { + const allKeys = Array.from(this.localCache.keys()); + + for (const key of allKeys) { + if (key.startsWith(this.getCacheKey(workspaceId, ''))) { + this.localCache.delete(key); + } + } + + return; + } + + for (const workspaceCacheKey of workspaceCacheKeys) { + const localKey = this.getCacheKey(workspaceId, workspaceCacheKey); + + this.localCache.delete(localKey); + } + } + + private partitionKeysByTTLStaleness( + workspaceId: string, + workspaceCacheKeys: string[], + ): { freshKeys: string[]; staleKeys: string[] } { + const freshKeys: string[] = []; + const staleKeys: string[] = []; + const now = Date.now(); + + for (const workspaceCacheKey of workspaceCacheKeys) { + const localKey = this.getCacheKey(workspaceId, workspaceCacheKey); + const cached = this.localCache.get(localKey); + + if ( + isDefined(cached) && + now - cached.lastCheckedAt < LOCAL_STALENESS_TTL_MS + ) { + freshKeys.push(workspaceCacheKey); + } else { + staleKeys.push(workspaceCacheKey); + } + } + + return { freshKeys, staleKeys }; + } + + private async resolveStaleKeys( + workspaceId: string, + workspaceCacheKeys: string[], + ): Promise> { + const result: Record = {}; + + const { validFromLocal, needsRedisCheck } = + await this.partitionKeysByLocalStaleness(workspaceId, workspaceCacheKeys); + + for (const workspaceCacheKey of validFromLocal) { + const localKey = this.getCacheKey(workspaceId, workspaceCacheKey); + const localEntry = this.localCache.get(localKey); + + if (!isDefined(localEntry)) { + continue; + } + + result[workspaceCacheKey] = localEntry.data; + this.updateLocalCache( + workspaceId, + workspaceCacheKey, + localEntry.data, + localEntry.hash, + ); + } + + if (needsRedisCheck.length === 0) { + return result; + } + + const { + validDataFromRedis: validFromRedis, + cacheKeysToRecomputeFromProviders: needsCompute, + } = await this.fetchDataFromRedis(workspaceId, needsRedisCheck); + + for (const { workspaceCacheKey, data, hash } of validFromRedis) { + result[workspaceCacheKey] = data; + this.updateLocalCache(workspaceId, workspaceCacheKey, data, hash); + } + + if (needsCompute.length === 0) { + return result; + } + + const computed = await this.computeAndStoreInRedis( + workspaceId, + needsCompute, + ); + + for (const { workspaceCacheKey, data, hash } of computed) { + result[workspaceCacheKey] = data; + this.updateLocalCache(workspaceId, workspaceCacheKey, data, hash); + } + + return result; + } + + private async partitionKeysByLocalStaleness( + workspaceId: string, + workspaceCacheKeys: string[], + ): Promise<{ + validFromLocal: string[]; + needsRedisCheck: string[]; + }> { + const validFromLocal: string[] = []; + const needsRedisCheck: string[] = []; + + const hashKeys = workspaceCacheKeys.map((workspaceCacheKey) => { + return `${workspaceId}:${workspaceCacheKey}:hash`; + }); + + const redisHashes = await this.cacheStorage.mget(hashKeys); + + for (let i = 0; i < workspaceCacheKeys.length; i++) { + const workspaceCacheKey = workspaceCacheKeys[i]; + const redisHash = redisHashes[i]; + const localKey = this.getCacheKey(workspaceId, workspaceCacheKey); + const localEntry = this.localCache.get(localKey); + + if ( + isDefined(localEntry) && + isDefined(redisHash) && + localEntry.hash === redisHash + ) { + validFromLocal.push(workspaceCacheKey); + } else { + needsRedisCheck.push(workspaceCacheKey); + } + } + + return { validFromLocal, needsRedisCheck }; + } + + private async fetchDataFromRedis( + workspaceId: string, + workspaceCacheKeys: string[], + ): Promise<{ + validDataFromRedis: Array<{ + workspaceCacheKey: string; + data: unknown; + hash: string; + }>; + cacheKeysToRecomputeFromProviders: string[]; + }> { + const validDataFromRedis: Array<{ + workspaceCacheKey: string; + data: unknown; + hash: string; + }> = []; + const cacheKeysToRecomputeFromProviders: string[] = []; + + const dataKeys = workspaceCacheKeys.map((workspaceCacheKey) => { + return `${this.getCacheKey(workspaceId, workspaceCacheKey)}:data`; + }); + + const redisData = await this.cacheStorage.mget(dataKeys); + + for (let i = 0; i < workspaceCacheKeys.length; i++) { + const workspaceCacheKey = workspaceCacheKeys[i]; + const data = redisData[i]; + + if (isDefined(data)) { + const hash = this.generateHash(data); + + validDataFromRedis.push({ workspaceCacheKey, data, hash }); + } else { + cacheKeysToRecomputeFromProviders.push(workspaceCacheKey); + } + } + + return { + validDataFromRedis, + cacheKeysToRecomputeFromProviders, + }; + } + + private getProviderOrThrow( + workspaceCacheKey: string, + ): WorkspaceCacheProvider { + const provider = this.workspaceCacheProviders.get(workspaceCacheKey); + + if (!isDefined(provider)) { + throw new Error( + `Cache provider with key "${workspaceCacheKey}" not found`, + ); + } + + return provider; + } + + private async computeAndStoreInRedis( + workspaceId: string, + workspaceCacheKeys: string[], + ): Promise< + Array<{ workspaceCacheKey: string; data: unknown; hash: string }> + > { + const computePromises = workspaceCacheKeys.map( + async (workspaceCacheKey) => { + const provider = this.getProviderOrThrow(workspaceCacheKey); + + const data = await provider.computeForCache(workspaceId); + + return { workspaceCacheKey, data }; + }, + ); + + const computed = await Promise.all(computePromises); + + const redisEntries = computed.flatMap(({ workspaceCacheKey, data }) => { + const hash = this.generateHash(data); + + return [ + { + key: `${this.getCacheKey(workspaceId, workspaceCacheKey)}:data`, + value: data, + }, + { + key: `${this.getCacheKey(workspaceId, workspaceCacheKey)}:hash`, + value: hash, + }, + ]; + }); + + await this.cacheStorage.mset(redisEntries); + + return computed.map(({ workspaceCacheKey, data }) => ({ + workspaceCacheKey, + data, + hash: this.generateHash(data), + })); + } + + private updateLocalCache( + workspaceId: string, + workspaceCacheKey: string, + data: unknown, + hash: string, + ): void { + const localKey = this.getCacheKey(workspaceId, workspaceCacheKey); + + this.localCache.set(localKey, { + data, + hash, + lastCheckedAt: Date.now(), + }); + } + + private getCacheKey(workspaceId: string, workspaceCacheKey: string): string { + return `${workspaceId}:${workspaceCacheKey}`; + } + + private generateHash(data: unknown): string { + return crypto + .createHash('sha256') + .update(JSON.stringify(data)) + .digest('hex'); + } +} diff --git a/packages/twenty-server/src/engine/workspace-cache/types/workspace-context-cache-entry.type.ts b/packages/twenty-server/src/engine/workspace-cache/types/workspace-context-cache-entry.type.ts new file mode 100644 index 00000000000..e051eafd53f --- /dev/null +++ b/packages/twenty-server/src/engine/workspace-cache/types/workspace-context-cache-entry.type.ts @@ -0,0 +1,5 @@ +export type WorkspaceContextLocalCacheEntry = { + data: T; + hash: string; + lastCheckedAt: number; +}; diff --git a/packages/twenty-server/src/engine/workspace-cache/types/workspace-context-data.type.ts b/packages/twenty-server/src/engine/workspace-cache/types/workspace-context-data.type.ts new file mode 100644 index 00000000000..9d552bb8fbf --- /dev/null +++ b/packages/twenty-server/src/engine/workspace-cache/types/workspace-context-data.type.ts @@ -0,0 +1,11 @@ +import { type ObjectsPermissionsByRoleId } from 'twenty-shared/types'; + +import { type FeatureFlagKey } from 'src/engine/core-modules/feature-flag/enums/feature-flag-key.enum'; +import { type ObjectMetadataMaps } from 'src/engine/metadata-modules/types/object-metadata-maps'; + +export type WorkspaceContextData = { + objectMetadataMaps: ObjectMetadataMaps; + metadataVersion: number; + featureFlagsMap: Record; + permissionsPerRoleId: ObjectsPermissionsByRoleId; +}; diff --git a/packages/twenty-server/src/engine/workspace-cache/workspace-cache-provider.service.ts b/packages/twenty-server/src/engine/workspace-cache/workspace-cache-provider.service.ts new file mode 100644 index 00000000000..5bb4553abe4 --- /dev/null +++ b/packages/twenty-server/src/engine/workspace-cache/workspace-cache-provider.service.ts @@ -0,0 +1,6 @@ +import { Injectable } from '@nestjs/common'; + +@Injectable() +export abstract class WorkspaceCacheProvider { + abstract computeForCache(workspaceId: string): Promise; +} diff --git a/packages/twenty-server/src/engine/workspace-cache/workspace-cache.module.ts b/packages/twenty-server/src/engine/workspace-cache/workspace-cache.module.ts new file mode 100644 index 00000000000..eb14fb2ebe2 --- /dev/null +++ b/packages/twenty-server/src/engine/workspace-cache/workspace-cache.module.ts @@ -0,0 +1,12 @@ +import { Module } from '@nestjs/common'; +import { DiscoveryModule } from '@nestjs/core'; + +import { CacheStorageModule } from 'src/engine/core-modules/cache-storage/cache-storage.module'; +import { WorkspaceCacheService } from 'src/engine/workspace-cache/services/workspace-cache.service'; + +@Module({ + imports: [CacheStorageModule, DiscoveryModule], + providers: [WorkspaceCacheService], + exports: [WorkspaceCacheService], +}) +export class WorkspaceContextCacheModule {}