houseassassin/twenty
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

Refactor workspace cache service (#16208)

Browse Source 
## Context
We've recently introduced a new workspace cache service which now acts
as a cache access and local storage for all workspace related data,
deprecating the individual specific services.
- Better performance through multiple caching/fetching strategies
- Consistent data access patterns across the codebase
- Reduced redis queries through MGET/MSET/PIPELINE with multiple cache
keys
This commit is contained in:
Weiko
2025-12-01 17:08:21 +01:00
committed by GitHub
parent 4e0545ebc5
commit 1eb2e44058
77 changed files with 1306 additions and 1863 deletions
Download Patch File Download Diff File Expand all files Collapse all files
packages/twenty-server/src/engine/workspace-cache/services/workspace-cache.service.ts
+176 -116
View File
@@ -5,30 +5,39 @@ import crypto from 'crypto';
import { isDefined } from 'twenty-shared/utils';
import { WorkspaceCacheProvider } from 'src/engine/workspace-cache/interfaces/workspace-cache-provider.service';
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';
import {
WORKSPACE_CACHE_KEYS_V2,
type WorkspaceCacheDataMap,
type WorkspaceCacheKeyName,
type WorkspaceCacheResult,
} from 'src/engine/workspace-cache/types/workspace-cache-key.type';
import { type WorkspaceLocalCacheEntry } from 'src/engine/workspace-cache/types/workspace-local-cache-entry.type';
const LOCAL_STALENESS_TTL_MS = 30_000;
const LOCAL_STALENESS_TTL_MS = 100;
const MEMOIZER_TTL_MS = 10_000;
type CacheDataType = WorkspaceCacheDataMap[WorkspaceCacheKeyName];
@Injectable()
export class WorkspaceCacheService implements OnModuleInit {
private readonly localCache = new Map<
string,
WorkspaceContextLocalCacheEntry<unknown>
WorkspaceLocalCacheEntry<CacheDataType>
>();
private readonly workspaceCacheProviders = new Map<
string,
WorkspaceCacheProvider<unknown>
WorkspaceCacheKeyName,
WorkspaceCacheProvider<CacheDataType>
>();
private readonly memoizer = new PromiseMemoizer<Record<string, unknown>>(
MEMOIZER_TTL_MS,
);
private readonly memoizer = new PromiseMemoizer<
Partial<WorkspaceCacheDataMap>
>(MEMOIZER_TTL_MS);
constructor(
@InjectCacheStorage(CacheStorageNamespace.EngineWorkspace)
@@ -47,41 +56,49 @@ export class WorkspaceCacheService implements OnModuleInit {
continue;
}
const workspaceCacheKey = this.reflector.get<string>(
const workspaceCacheKeyName = this.reflector.get<WorkspaceCacheKeyName>(
WORKSPACE_CACHE_KEY,
instance.constructor,
);
if (
isDefined(workspaceCacheKey) &&
isDefined(workspaceCacheKeyName) &&
instance instanceof WorkspaceCacheProvider
) {
this.workspaceCacheProviders.set(workspaceCacheKey, instance);
this.workspaceCacheProviders.set(workspaceCacheKeyName, instance);
}
}
}
async getOrRecompute<
T extends Record<string, unknown> = Record<string, unknown>,
>(workspaceId: string, workspaceCacheKeys: string[]): Promise<T> {
async getOrRecompute<const K extends WorkspaceCacheKeyName[]>(
workspaceId: string,
workspaceCacheKeyNames: K,
): Promise<WorkspaceCacheResult<K>> {
const memoKey =
`${workspaceId}-${[...workspaceCacheKeys].sort().join(',')}` as const;
`${workspaceId}-${[...workspaceCacheKeyNames].sort().join(',')}` as const;
const result = await this.memoizer.memoizePromiseAndExecute(
memoKey,
async () => {
const freshResult: Record<string, unknown> = {};
const freshResult: Partial<WorkspaceCacheDataMap> = {};
const { freshKeys, staleKeys } = this.partitionKeysByTTLStaleness(
workspaceId,
workspaceCacheKeys,
workspaceCacheKeyNames,
);
for (const workspaceCacheKey of freshKeys) {
const localKey = this.getCacheKey(workspaceId, workspaceCacheKey);
for (const workspaceCacheKeyName of freshKeys) {
const localKey = this.buildCacheKey(
workspaceId,
workspaceCacheKeyName,
);
const cached = this.localCache.get(localKey);
freshResult[workspaceCacheKey] = cached?.data;
if (isDefined(cached)) {
Object.assign(freshResult, {
[workspaceCacheKeyName]: cached.data,
});
}
}
if (staleKeys.length === 0) {
@@ -97,48 +114,84 @@ export class WorkspaceCacheService implements OnModuleInit {
},
);
return result as T;
return result as WorkspaceCacheResult<K>;
}
invalidate(workspaceId: string, workspaceCacheKeys?: string[]): void {
if (!isDefined(workspaceCacheKeys) || workspaceCacheKeys.length === 0) {
const allKeys = Array.from(this.localCache.keys());
async invalidateAndRecompute(
workspaceId: string,
workspaceCacheKeys: WorkspaceCacheKeyName[],
): Promise<void> {
await this.memoizer.clearKeys(`${workspaceId}-`);
for (const key of allKeys) {
if (key.startsWith(this.getCacheKey(workspaceId, ''))) {
this.localCache.delete(key);
}
}
await this.flush(workspaceId, workspaceCacheKeys);
await this.recomputeCache(workspaceId, workspaceCacheKeys);
}
return;
}
async flush(
workspaceId: string,
workspaceCacheKeys: WorkspaceCacheKeyName[],
): Promise<void> {
await this.deleteFromRedis(workspaceId, workspaceCacheKeys);
this.deleteFromLocalCache(workspaceId, workspaceCacheKeys);
}
for (const workspaceCacheKey of workspaceCacheKeys) {
const localKey = this.getCacheKey(workspaceId, workspaceCacheKey);
private deleteFromLocalCache(
workspaceId: string,
workspaceCacheKeys: WorkspaceCacheKeyName[],
): void {
for (const workspaceCacheKeyName of workspaceCacheKeys) {
const localKey = this.buildCacheKey(workspaceId, workspaceCacheKeyName);
this.localCache.delete(localKey);
}
}
private partitionKeysByTTLStaleness(
private async deleteFromRedis(
workspaceId: string,
workspaceCacheKeys: string[],
): { freshKeys: string[]; staleKeys: string[] } {
const freshKeys: string[] = [];
const staleKeys: string[] = [];
workspaceCacheKeys: WorkspaceCacheKeyName[],
): Promise<void> {
const keysToDelete = workspaceCacheKeys.flatMap((workspaceCacheKeyName) => {
const baseKey = this.buildCacheKey(workspaceId, workspaceCacheKeyName);
return [`${baseKey}:data`, `${baseKey}:hash`];
});
await this.cacheStorage.mdel(keysToDelete);
}
private async recomputeCache(
workspaceId: string,
workspaceCacheKeys: WorkspaceCacheKeyName[],
): Promise<void> {
const computed = await this.computeAndStoreInRedis(
workspaceId,
workspaceCacheKeys,
);
for (const { workspaceCacheKeyName, data, hash } of computed) {
this.setInLocalCache(workspaceId, workspaceCacheKeyName, data, hash);
}
}
private partitionKeysByTTLStaleness<K extends WorkspaceCacheKeyName>(
workspaceId: string,
workspaceCacheKeys: readonly K[],
): { freshKeys: K[]; staleKeys: K[] } {
const freshKeys: K[] = [];
const staleKeys: K[] = [];
const now = Date.now();
for (const workspaceCacheKey of workspaceCacheKeys) {
const localKey = this.getCacheKey(workspaceId, workspaceCacheKey);
for (const workspaceCacheKeyName of workspaceCacheKeys) {
const localKey = this.buildCacheKey(workspaceId, workspaceCacheKeyName);
const cached = this.localCache.get(localKey);
if (
isDefined(cached) &&
now - cached.lastCheckedAt < LOCAL_STALENESS_TTL_MS
) {
freshKeys.push(workspaceCacheKey);
freshKeys.push(workspaceCacheKeyName);
} else {
staleKeys.push(workspaceCacheKey);
staleKeys.push(workspaceCacheKeyName);
}
}
@@ -147,25 +200,25 @@ export class WorkspaceCacheService implements OnModuleInit {
private async resolveStaleKeys(
workspaceId: string,
workspaceCacheKeys: string[],
): Promise<Record<string, unknown>> {
const result: Record<string, unknown> = {};
workspaceCacheKeys: WorkspaceCacheKeyName[],
): Promise<Partial<WorkspaceCacheDataMap>> {
const result: Partial<WorkspaceCacheDataMap> = {};
const { validFromLocal, needsRedisCheck } =
await this.partitionKeysByLocalStaleness(workspaceId, workspaceCacheKeys);
for (const workspaceCacheKey of validFromLocal) {
const localKey = this.getCacheKey(workspaceId, workspaceCacheKey);
for (const workspaceCacheKeyName of validFromLocal) {
const localKey = this.buildCacheKey(workspaceId, workspaceCacheKeyName);
const localEntry = this.localCache.get(localKey);
if (!isDefined(localEntry)) {
continue;
}
result[workspaceCacheKey] = localEntry.data;
this.updateLocalCache(
Object.assign(result, { [workspaceCacheKeyName]: localEntry.data });
this.setInLocalCache(
workspaceId,
workspaceCacheKey,
workspaceCacheKeyName,
localEntry.data,
localEntry.hash,
);
@@ -180,9 +233,9 @@ export class WorkspaceCacheService implements OnModuleInit {
cacheKeysToRecomputeFromProviders: needsCompute,
} = await this.fetchDataFromRedis(workspaceId, needsRedisCheck);
for (const { workspaceCacheKey, data, hash } of validFromRedis) {
result[workspaceCacheKey] = data;
this.updateLocalCache(workspaceId, workspaceCacheKey, data, hash);
for (const { workspaceCacheKeyName, data, hash } of validFromRedis) {
Object.assign(result, { [workspaceCacheKeyName]: data });
this.setInLocalCache(workspaceId, workspaceCacheKeyName, data, hash);
}
if (needsCompute.length === 0) {
@@ -194,9 +247,9 @@ export class WorkspaceCacheService implements OnModuleInit {
needsCompute,
);
for (const { workspaceCacheKey, data, hash } of computed) {
result[workspaceCacheKey] = data;
this.updateLocalCache(workspaceId, workspaceCacheKey, data, hash);
for (const { workspaceCacheKeyName, data, hash } of computed) {
Object.assign(result, { [workspaceCacheKeyName]: data });
this.setInLocalCache(workspaceId, workspaceCacheKeyName, data, hash);
}
return result;
@@ -204,24 +257,24 @@ export class WorkspaceCacheService implements OnModuleInit {
private async partitionKeysByLocalStaleness(
workspaceId: string,
workspaceCacheKeys: string[],
workspaceCacheKeys: WorkspaceCacheKeyName[],
): Promise<{
validFromLocal: string[];
needsRedisCheck: string[];
validFromLocal: WorkspaceCacheKeyName[];
needsRedisCheck: WorkspaceCacheKeyName[];
}> {
const validFromLocal: string[] = [];
const needsRedisCheck: string[] = [];
const validFromLocal: WorkspaceCacheKeyName[] = [];
const needsRedisCheck: WorkspaceCacheKeyName[] = [];
const hashKeys = workspaceCacheKeys.map((workspaceCacheKey) => {
return `${workspaceId}:${workspaceCacheKey}:hash`;
});
const hashKeys = workspaceCacheKeys.map(
(workspaceCacheKeyName) =>
`${this.buildCacheKey(workspaceId, workspaceCacheKeyName)}:hash`,
);
const redisHashes = await this.cacheStorage.mget<string>(hashKeys);
for (let i = 0; i < workspaceCacheKeys.length; i++) {
const workspaceCacheKey = workspaceCacheKeys[i];
const redisHash = redisHashes[i];
const localKey = this.getCacheKey(workspaceId, workspaceCacheKey);
for (const [index, workspaceCacheKeyName] of workspaceCacheKeys.entries()) {
const redisHash = redisHashes[index];
const localKey = this.buildCacheKey(workspaceId, workspaceCacheKeyName);
const localEntry = this.localCache.get(localKey);
if (
@@ -229,9 +282,9 @@ export class WorkspaceCacheService implements OnModuleInit {
isDefined(redisHash) &&
localEntry.hash === redisHash
) {
validFromLocal.push(workspaceCacheKey);
validFromLocal.push(workspaceCacheKeyName);
} else {
needsRedisCheck.push(workspaceCacheKey);
needsRedisCheck.push(workspaceCacheKeyName);
}
}
@@ -240,38 +293,38 @@ export class WorkspaceCacheService implements OnModuleInit {
private async fetchDataFromRedis(
workspaceId: string,
workspaceCacheKeys: string[],
workspaceCacheKeys: WorkspaceCacheKeyName[],
): Promise<{
validDataFromRedis: Array<{
workspaceCacheKey: string;
data: unknown;
workspaceCacheKeyName: WorkspaceCacheKeyName;
data: CacheDataType;
hash: string;
}>;
cacheKeysToRecomputeFromProviders: string[];
cacheKeysToRecomputeFromProviders: WorkspaceCacheKeyName[];
}> {
const validDataFromRedis: Array<{
workspaceCacheKey: string;
data: unknown;
workspaceCacheKeyName: WorkspaceCacheKeyName;
data: CacheDataType;
hash: string;
}> = [];
const cacheKeysToRecomputeFromProviders: string[] = [];
const cacheKeysToRecomputeFromProviders: WorkspaceCacheKeyName[] = [];
const dataKeys = workspaceCacheKeys.map((workspaceCacheKey) => {
return `${this.getCacheKey(workspaceId, workspaceCacheKey)}:data`;
});
const dataKeys = workspaceCacheKeys.map(
(workspaceCacheKeyName) =>
`${this.buildCacheKey(workspaceId, workspaceCacheKeyName)}:data`,
);
const redisData = await this.cacheStorage.mget<unknown>(dataKeys);
const redisData = await this.cacheStorage.mget<CacheDataType>(dataKeys);
for (let i = 0; i < workspaceCacheKeys.length; i++) {
const workspaceCacheKey = workspaceCacheKeys[i];
const data = redisData[i];
for (const [index, workspaceCacheKeyName] of workspaceCacheKeys.entries()) {
const data = redisData[index];
if (isDefined(data)) {
const hash = this.generateHash(data);
validDataFromRedis.push({ workspaceCacheKey, data, hash });
validDataFromRedis.push({ workspaceCacheKeyName, data, hash });
} else {
cacheKeysToRecomputeFromProviders.push(workspaceCacheKey);
cacheKeysToRecomputeFromProviders.push(workspaceCacheKeyName);
}
}
@@ -282,13 +335,13 @@ export class WorkspaceCacheService implements OnModuleInit {
}
private getProviderOrThrow(
workspaceCacheKey: string,
): WorkspaceCacheProvider<unknown> {
const provider = this.workspaceCacheProviders.get(workspaceCacheKey);
workspaceCacheKeyName: WorkspaceCacheKeyName,
): WorkspaceCacheProvider<CacheDataType> {
const provider = this.workspaceCacheProviders.get(workspaceCacheKeyName);
if (!isDefined(provider)) {
throw new Error(
`Cache provider with key "${workspaceCacheKey}" not found`,
`Cache provider with key name "${workspaceCacheKeyName}" not found`,
);
}
@@ -297,53 +350,57 @@ export class WorkspaceCacheService implements OnModuleInit {
private async computeAndStoreInRedis(
workspaceId: string,
workspaceCacheKeys: string[],
workspaceCacheKeys: WorkspaceCacheKeyName[],
): Promise<
Array<{ workspaceCacheKey: string; data: unknown; hash: string }>
Array<{
workspaceCacheKeyName: WorkspaceCacheKeyName;
data: CacheDataType;
hash: string;
}>
> {
const computePromises = workspaceCacheKeys.map(
async (workspaceCacheKey) => {
const provider = this.getProviderOrThrow(workspaceCacheKey);
async (workspaceCacheKeyName) => {
const provider = this.getProviderOrThrow(workspaceCacheKeyName);
const data = await provider.computeForCache(workspaceId);
return { workspaceCacheKey, data };
return { workspaceCacheKeyName, data };
},
);
const computed = await Promise.all(computePromises);
const redisEntries = computed.flatMap(({ workspaceCacheKey, data }) => {
const redisEntries: Array<{ key: string; value: unknown }> = [];
for (const { workspaceCacheKeyName, data } of computed) {
const hash = this.generateHash(data);
return [
{
key: `${this.getCacheKey(workspaceId, workspaceCacheKey)}:data`,
value: data,
},
{
key: `${this.getCacheKey(workspaceId, workspaceCacheKey)}:hash`,
value: hash,
},
];
});
redisEntries.push({
key: `${this.buildCacheKey(workspaceId, workspaceCacheKeyName)}:data`,
value: data,
});
redisEntries.push({
key: `${this.buildCacheKey(workspaceId, workspaceCacheKeyName)}:hash`,
value: hash,
});
}
await this.cacheStorage.mset(redisEntries);
return computed.map(({ workspaceCacheKey, data }) => ({
workspaceCacheKey,
return computed.map(({ workspaceCacheKeyName, data }) => ({
workspaceCacheKeyName,
data,
hash: this.generateHash(data),
}));
}
private updateLocalCache(
private setInLocalCache(
workspaceId: string,
workspaceCacheKey: string,
data: unknown,
workspaceCacheKeyName: WorkspaceCacheKeyName,
data: CacheDataType,
hash: string,
): void {
const localKey = this.getCacheKey(workspaceId, workspaceCacheKey);
const localKey = this.buildCacheKey(workspaceId, workspaceCacheKeyName);
this.localCache.set(localKey, {
data,
@@ -352,8 +409,11 @@ export class WorkspaceCacheService implements OnModuleInit {
});
}
private getCacheKey(workspaceId: string, workspaceCacheKey: string): string {
return `${workspaceId}:${workspaceCacheKey}`;
private buildCacheKey(
workspaceId: string,
workspaceCacheKeyName: WorkspaceCacheKeyName,
): string {
return `${WORKSPACE_CACHE_KEYS_V2[workspaceCacheKeyName]}:${workspaceId}`;
}
private generateHash(data: unknown): string {