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 {}