feat: implement tag-based cache invalidation for related asset data

apps/api/src/app/redis-cache/redis-cache.service.ts

@@ -135,11 +135,54 @@ export class RedisCacheService {
     return this.cache.clear();
   }
 
-  public async set(key: string, value: string, ttl?: number) {
+  public async set(
-    return this.cache.set(
+    key: string,
-      key,
+    value: string,
-      value,
+    optionsOrTtl?: number | { ttl?: number; tags?: string[] }
-      ttl ?? this.configurationService.get('CACHE_TTL')
+  ) {
+    const ttl =
+      typeof optionsOrTtl === 'number'
+        ? optionsOrTtl
+        : optionsOrTtl?.ttl ?? this.configurationService.get('CACHE_TTL');
+
+    const tags =
+      typeof optionsOrTtl === 'object' ? optionsOrTtl.tags ?? [] : [];
+
+    const result = await this.cache.set(key, value, ttl);
+
+    if (tags.length > 0) {
+      const redisClient = this.getRedisClient();
+
+      if (redisClient && typeof redisClient.sadd === 'function') {
+        for (const tag of tags) {
+          await redisClient.sadd(`tag:${tag}`, key);
+        }
+      }
+    }
+
+    return result;
+  }
+
+  public async invalidateByTag(tag: string) {
+    const redisClient = this.getRedisClient();
+
+    if (redisClient && typeof redisClient.smembers === 'function') {
+      const keys = (await redisClient.smembers(`tag:${tag}`)) as string[];
+
+      if (keys?.length > 0) {
+        await this.cache.mdel(keys);
+      }
+
+      await redisClient.del(`tag:${tag}`);
+    }
+  }
+
+  private getRedisClient() {
+    return (
+      (this.client as any).opts?.store?.redis ||
+      (this.client as any).opts?.store?.client ||
+      (this.client as any).client ||
+      (this.client as any).redis
     );
   }
 

apps/api/src/services/queues/portfolio-snapshot/portfolio-snapshot.processor.ts

@@ -85,6 +85,15 @@ export class PortfolioSnapshotProcessor {
         this.configurationService.get('CACHE_QUOTES_TTL')
       );
 
+      const tags = [
+        `portfolio:${job.data.userId}`,
+        ...new Set(
+          snapshot.positions
+            .filter((p) => p.symbol)
+            .map((p) => `symbol:${p.symbol}`)
+        )
+      ];
+
       await this.redisCacheService.set(
         this.redisCacheService.getPortfolioSnapshotKey({
           calculationType: job.data.calculationType,
@@ -95,7 +104,7 @@ export class PortfolioSnapshotProcessor {
           expiration: expiration.getTime(),
           portfolioSnapshot: snapshot
         } as unknown as PortfolioSnapshotValue),
-        CACHE_TTL_INFINITE
+        { ttl: CACHE_TTL_INFINITE, tags }
       );
 
       return snapshot;