feat(api, application-generic): enhance workflow preferences retrieval with batch processing and caching improvements (#11646)

Author: merrcury

apps/api/src/app/subscribers/usecases/get-subscriber-preference/get-subscriber-preference.usecase.ts

@@ -40,7 +40,8 @@ export class GetSubscriberPreference {
     private notificationTemplateRepository: NotificationTemplateRepository,
     private preferencesRepository: PreferencesRepository,
     private featureFlagsService: FeatureFlagsService,
-    private inMemoryLRUCacheService: InMemoryLRUCacheService
+    private inMemoryLRUCacheService: InMemoryLRUCacheService,
+    private getPreferences: GetPreferences
   ) {}
 
   @InstrumentUsecase()
@@ -296,27 +297,16 @@ export class GetSubscriberPreference {
           );
 
     const [
-      workflowResourcePreferences,
-      workflowUserPreferences,
+      { workflowResourcePreferences, workflowUserPreferences },
       subscriberWorkflowPreferences,
       subscriberGlobalPreferences,
     ] = await Promise.all([
-      this.preferencesRepository.findForComputation(
-        {
-          ...baseQuery,
-          _templateId: { $in: workflowIds },
-          type: PreferencesTypeEnum.WORKFLOW_RESOURCE,
-        },
-        readOptions
-      ),
-      this.preferencesRepository.findForComputation(
-        {
-          ...baseQuery,
-          _templateId: { $in: workflowIds },
-          type: PreferencesTypeEnum.USER_WORKFLOW,
-        },
-        readOptions
-      ),
+      this.getWorkflowLevelPreferences({
+        environmentId,
+        organizationId,
+        workflowIds,
+        readOptions,
+      }),
       this.preferencesRepository.findForComputation(
         {
           ...baseQuery,
@@ -338,6 +328,52 @@ export class GetSubscriberPreference {
     };
   }
 
+  /**
+   * Resolves the subscriber-independent workflow-level preferences (WORKFLOW_RESOURCE and
+   * USER_WORKFLOW) for a set of workflows by delegating to the canonical, in-flight-coalesced
+   * `GetPreferences.getWorkflowPreferencesByIds` reader (shared with the single-workflow path),
+   * then flattens the per-workflow tuples into the two arrays the merge step consumes.
+   *
+   * The returned entries are shared cache references and must be treated as immutable; the merge
+   * pipeline (`MergePreferences`) only reads them and produces fresh objects.
+   */
+  @Instrument()
+  private async getWorkflowLevelPreferences({
+    environmentId,
+    organizationId,
+    workflowIds,
+    readOptions,
+  }: {
+    environmentId: string;
+    organizationId: string;
+    workflowIds: string[];
+    readOptions: { readPreference: 'secondaryPreferred' | 'primary' };
+  }): Promise<{
+    workflowResourcePreferences: PreferencesEntity[];
+    workflowUserPreferences: PreferencesEntity[];
+  }> {
+    const tuplesByWorkflowId = await this.getPreferences.getWorkflowPreferencesByIds({
+      environmentId,
+      organizationId,
+      workflowIds,
+      readOptions,
+    });
+
+    const workflowResourcePreferences: PreferencesEntity[] = [];
+    const workflowUserPreferences: PreferencesEntity[] = [];
+
+    for (const [workflowResourcePreference, workflowUserPreference] of tuplesByWorkflowId.values()) {
+      if (workflowResourcePreference) {
+        workflowResourcePreferences.push(workflowResourcePreference);
+      }
+      if (workflowUserPreference) {
+        workflowUserPreferences.push(workflowUserPreference);
+      }
+    }
+
+    return { workflowResourcePreferences, workflowUserPreferences };
+  }
+
   @Instrument()
   private async getActiveWorkflows({
     organizationId,

libs/application-generic/src/services/in-memory-lru-cache/in-memory-lru-cache.service.spec.ts

@@ -249,6 +249,144 @@ describe('InMemoryLRUCacheService', () => {
     });
   });
 
+  describe('getMany', () => {
+    afterEach(() => {
+      service.invalidateAll(InMemoryLRUCacheStore.WORKFLOW_PREFERENCES);
+    });
+
+    const buildFetchMissing = () =>
+      jest.fn(async (missingKeys: string[]) => {
+        const fetched = new Map<string, any>();
+        for (const key of missingKeys) {
+          fetched.set(key, [{ id: `resource-${key}` }, { id: `user-${key}` }]);
+        }
+
+        return fetched;
+      });
+
+    it('returns an empty map without fetching when given no keys', async () => {
+      const fetchMissing = buildFetchMissing();
+
+      const result = await service.getMany(InMemoryLRUCacheStore.WORKFLOW_PREFERENCES, [], fetchMissing, {
+        environmentId: 'env1',
+      });
+
+      expect(result.size).toBe(0);
+      expect(fetchMissing).not.toHaveBeenCalled();
+    });
+
+    it('fetches all missing keys in a single batch call and caches them', async () => {
+      featureFlagsService.getFlag.mockResolvedValue(true);
+      const fetchMissing = buildFetchMissing();
+
+      const result = await service.getMany(
+        InMemoryLRUCacheStore.WORKFLOW_PREFERENCES,
+        ['env1:wf_1', 'env1:wf_2'],
+        fetchMissing,
+        { environmentId: 'env1', organizationId: 'org1' }
+      );
+
+      expect(fetchMissing).toHaveBeenCalledTimes(1);
+      expect(fetchMissing).toHaveBeenCalledWith(['env1:wf_1', 'env1:wf_2']);
+      expect(featureFlagsService.getFlag).toHaveBeenCalledWith(
+        expect.objectContaining({ key: FeatureFlagsKeysEnum.IS_LRU_CACHE_ENABLED })
+      );
+      expect(result.get('env1:wf_1')).toEqual([{ id: 'resource-env1:wf_1' }, { id: 'user-env1:wf_1' }]);
+      expect(result.get('env1:wf_2')).toEqual([{ id: 'resource-env1:wf_2' }, { id: 'user-env1:wf_2' }]);
+    });
+
+    it('serves cache hits without calling fetchMissing again', async () => {
+      featureFlagsService.getFlag.mockResolvedValue(true);
+      const fetchMissing = buildFetchMissing();
+
+      await service.getMany(InMemoryLRUCacheStore.WORKFLOW_PREFERENCES, ['env1:wf_1'], fetchMissing, {
+        environmentId: 'env1',
+      });
+      const result = await service.getMany(InMemoryLRUCacheStore.WORKFLOW_PREFERENCES, ['env1:wf_1'], fetchMissing, {
+        environmentId: 'env1',
+      });
+
+      expect(fetchMissing).toHaveBeenCalledTimes(1);
+      expect(result.get('env1:wf_1')).toEqual([{ id: 'resource-env1:wf_1' }, { id: 'user-env1:wf_1' }]);
+    });
+
+    it('only fetches the missing keys when the cache is partially warm', async () => {
+      featureFlagsService.getFlag.mockResolvedValue(true);
+      const fetchMissing = buildFetchMissing();
+
+      await service.getMany(InMemoryLRUCacheStore.WORKFLOW_PREFERENCES, ['env1:wf_1'], fetchMissing, {
+        environmentId: 'env1',
+      });
+      const result = await service.getMany(
+        InMemoryLRUCacheStore.WORKFLOW_PREFERENCES,
+        ['env1:wf_1', 'env1:wf_2'],
+        fetchMissing,
+        { environmentId: 'env1' }
+      );
+
+      expect(fetchMissing).toHaveBeenCalledTimes(2);
+      expect(fetchMissing).toHaveBeenNthCalledWith(2, ['env1:wf_2']);
+      expect(result.size).toBe(2);
+    });
+
+    it('coalesces concurrent callers for the same key into a single fetch', async () => {
+      featureFlagsService.getFlag.mockResolvedValue(true);
+      let fetchCount = 0;
+      const fetchMissing = jest.fn(
+        (missingKeys: string[]) =>
+          new Promise<Map<string, any>>((resolve) => {
+            setTimeout(() => {
+              fetchCount++;
+              const fetched = new Map<string, any>();
+              for (const key of missingKeys) {
+                fetched.set(key, [{ id: `resource-${fetchCount}` }, null]);
+              }
+              resolve(fetched);
+            }, 10);
+          })
+      );
+
+      const [result1, result2, result3] = await Promise.all([
+        service.getMany(InMemoryLRUCacheStore.WORKFLOW_PREFERENCES, ['env1:wf_1'], fetchMissing, {
+          environmentId: 'env1',
+        }),
+        service.getMany(InMemoryLRUCacheStore.WORKFLOW_PREFERENCES, ['env1:wf_1'], fetchMissing, {
+          environmentId: 'env1',
+        }),
+        service.getMany(InMemoryLRUCacheStore.WORKFLOW_PREFERENCES, ['env1:wf_1'], fetchMissing, {
+          environmentId: 'env1',
+        }),
+      ]);
+
+      expect(fetchMissing).toHaveBeenCalledTimes(1);
+      expect(result1.get('env1:wf_1')).toEqual([{ id: 'resource-1' }, null]);
+      expect(result2.get('env1:wf_1')).toEqual([{ id: 'resource-1' }, null]);
+      expect(result3.get('env1:wf_1')).toEqual([{ id: 'resource-1' }, null]);
+    });
+
+    it('bypasses the cache and fetches every key when the feature flag is disabled', async () => {
+      featureFlagsService.getFlag.mockResolvedValue(false);
+      const fetchMissing = buildFetchMissing();
+
+      const result = await service.getMany(
+        InMemoryLRUCacheStore.WORKFLOW_PREFERENCES,
+        ['env1:wf_1', 'env1:wf_2'],
+        fetchMissing,
+        { environmentId: 'env1' }
+      );
+
+      expect(fetchMissing).toHaveBeenCalledTimes(1);
+      expect(fetchMissing).toHaveBeenCalledWith(['env1:wf_1', 'env1:wf_2']);
+      expect(result.size).toBe(2);
+
+      await service.getMany(InMemoryLRUCacheStore.WORKFLOW_PREFERENCES, ['env1:wf_1', 'env1:wf_2'], fetchMissing, {
+        environmentId: 'env1',
+      });
+
+      expect(fetchMissing).toHaveBeenCalledTimes(2);
+    });
+  });
+
   describe('getIfCached', () => {
     it('should return undefined for non-existent key', () => {
       const result = service.getIfCached(InMemoryLRUCacheStore.WORKFLOW, 'nonexistent');

libs/application-generic/src/services/in-memory-lru-cache/in-memory-lru-cache.service.ts

@@ -6,7 +6,14 @@ import { CacheStoreDataTypeMap, InMemoryLRUCacheStore, STORE_CONFIGS, StoreConfi
 
 type EntityStore<T = unknown> = {
   cache: LRUCache<string, T>;
-  inflightRequests: Map<string, Promise<T>>;
+  /**
+   * In-flight fetches keyed by effective cache key, shared by `get()` and `getMany()` so
+   * concurrent callers coalesce onto one fetch. The resolved value is `T | undefined` because
+   * `getMany()` registers a promise per requested key and a batch fetch may legitimately omit a
+   * key (resolving `undefined`); `get()` treats an `undefined` resolution as a miss and falls
+   * back to its own fetch, so the two methods can safely share a store.
+   */
+  inflightRequests: Map<string, Promise<T | undefined>>;
   config: StoreConfig;
 };
 
@@ -45,7 +52,11 @@ export class InMemoryLRUCacheService {
 
     const inflightRequest = store.inflightRequests.get(effectiveKey);
     if (inflightRequest) {
-      return inflightRequest;
+      const inflightResult = await inflightRequest;
+      if (inflightResult !== undefined) {
+        return inflightResult;
+      }
+      // A shared `getMany()` batch had no value for this key — fall through to our own fetch.
     }
 
     const fetchPromise = fetchFn()
@@ -65,6 +76,93 @@ export class InMemoryLRUCacheService {
     return fetchPromise;
   }
 
+  /**
+   * Batch variant of `get()` that resolves many keys at once while preserving the same
+   * caching and in-flight coalescing semantics per key:
+   * - cache hits are served from memory,
+   * - keys with an in-flight fetch (from this or a concurrent call) reuse that promise,
+   * - only the remaining keys are passed to `fetchMissing` in a single call.
+   *
+   * This keeps the cold-cache/TTL-expiry path stampede-safe (K concurrent callers for the same
+   * key set trigger one `fetchMissing` per key, not K), which a manual `getIfCached()` + `set()`
+   * loop would lose. `fetchMissing` is expected to return an entry for every requested key.
+   */
+  async getMany<TStore extends InMemoryLRUCacheStore>(
+    storeName: TStore,
+    keys: string[],
+    fetchMissing: (missingKeys: string[]) => Promise<Map<string, CacheStoreDataTypeMap[TStore]>>,
+    opts?: GetOptions
+  ): Promise<Map<string, CacheStoreDataTypeMap[TStore]>> {
+    const result = new Map<string, CacheStoreDataTypeMap[TStore]>();
+
+    if (keys.length === 0) {
+      return result;
+    }
+
+    const store = this.getOrCreateStore<CacheStoreDataTypeMap[TStore]>(storeName);
+    const isCacheEnabled = await this.isCacheEnabled(store.config, opts);
+
+    if (!isCacheEnabled || opts?.skipCache) {
+      return fetchMissing(keys);
+    }
+
+    const pending: Array<{ key: string; promise: Promise<CacheStoreDataTypeMap[TStore] | undefined> }> = [];
+    const missingKeys: string[] = [];
+
+    for (const key of keys) {
+      const effectiveKey = this.resolveKey(key, opts?.cacheVariant);
+
+      const cached = store.cache.get(effectiveKey);
+      if (cached !== undefined) {
+        result.set(key, cached);
+        continue;
+      }
+
+      const inflightRequest = store.inflightRequests.get(effectiveKey);
+      if (inflightRequest) {
+        pending.push({ key, promise: inflightRequest });
+        continue;
+      }
+
+      missingKeys.push(key);
+    }
+
+    if (missingKeys.length > 0) {
+      const batchPromise = fetchMissing(missingKeys);
+
+      for (const key of missingKeys) {
+        const effectiveKey = this.resolveKey(key, opts?.cacheVariant);
+
+        const perKeyPromise = batchPromise
+          .then((fetched) => {
+            const value = fetched.get(key);
+            if (value !== null && value !== undefined) {
+              store.cache.set(effectiveKey, value);
+            }
+
+            return value;
+          })
+          .finally(() => {
+            store.inflightRequests.delete(effectiveKey);
+          });
+
+        store.inflightRequests.set(effectiveKey, perKeyPromise);
+        pending.push({ key, promise: perKeyPromise });
+      }
+    }
+
+    await Promise.all(
+      pending.map(async ({ key, promise }) => {
+        const value = await promise;
+        if (value !== undefined) {
+          result.set(key, value);
+        }
+      })
+    );
+
+    return result;
+  }
+
   getIfCached<TStore extends InMemoryLRUCacheStore>(
     storeName: TStore,
     key: string
@@ -124,7 +222,7 @@ export class InMemoryLRUCacheService {
           max: config.max,
           ttl: config.ttl,
         }),
-        inflightRequests: new Map<string, Promise<T>>(),
+        inflightRequests: new Map<string, Promise<T | undefined>>(),
         config,
       };
       STORES.set(storeName, store as EntityStore);