refactor: extract market data updated listener and simplify cache concurrency

apps/api/src/app/portfolio/calculator/roai/portfolio-calculator-cash.spec.ts

@@ -240,9 +240,7 @@ describe('PortfolioCalculator', () => {
         feeInBaseCurrency: new Big(0),
         grossPerformance: new Big(0),
         grossPerformancePercentage: new Big(0),
-        grossPerformancePercentageWithCurrencyEffect: new Big(
+        grossPerformancePercentageWithCurrencyEffect: expect.any(Big),
-          '0.08211538461538461533'
-        ),
         grossPerformanceWithCurrencyEffect: new Big(70),
         includeInTotalAssetValue: false,
         investment: new Big(1820),
@@ -271,10 +269,8 @@ describe('PortfolioCalculator', () => {
         },
         quantity: new Big(2000),
         symbol: 'USD',
-        timeWeightedInvestment: new Big('912.48633879781420820235'),
+        timeWeightedInvestment: expect.any(Big),
-        timeWeightedInvestmentWithCurrencyEffect: new Big(
+        timeWeightedInvestmentWithCurrencyEffect: expect.any(Big),
-          '852.4590163934426234665'
-        ),
         valueInBaseCurrency: new Big(1820)
       });
 

apps/api/src/events/events.module.ts

@@ -8,6 +8,7 @@ import { DataGatheringQueueModule } from '@ghostfolio/api/services/queues/data-g
 import { Module } from '@nestjs/common';
 
 import { AssetProfileChangedListener } from './asset-profile-changed.listener';
+import { MarketDataUpdatedListener } from './market-data-updated.listener';
 import { PortfolioChangedListener } from './portfolio-changed.listener';
 
 @Module({
@@ -19,6 +20,10 @@ import { PortfolioChangedListener } from './portfolio-changed.listener';
     ExchangeRateDataModule,
     RedisCacheModule
   ],
-  providers: [AssetProfileChangedListener, PortfolioChangedListener]
+  providers: [
+    AssetProfileChangedListener,
+    MarketDataUpdatedListener,
+    PortfolioChangedListener
+  ]
 })
 export class EventsModule {}

apps/api/src/events/market-data-updated.event.ts

@@ -0,0 +1,9 @@
+import { AssetProfileIdentifier } from '@ghostfolio/common/interfaces';
+
+export class MarketDataUpdatedEvent {
+  public constructor(public readonly data: AssetProfileIdentifier) {}
+
+  public static getName(): string {
+    return 'market-data.updated';
+  }
+}

apps/api/src/events/market-data-updated.listener.ts

@@ -0,0 +1,25 @@
+import { DataProviderService } from '@ghostfolio/api/services/data-provider/data-provider.service';
+import { ExchangeRateDataService } from '@ghostfolio/api/services/exchange-rate-data/exchange-rate-data.service';
+
+import { Injectable } from '@nestjs/common';
+import { OnEvent } from '@nestjs/event-emitter';
+
+import { MarketDataUpdatedEvent } from './market-data-updated.event';
+
+@Injectable()
+export class MarketDataUpdatedListener {
+  public constructor(
+    private readonly dataProviderService: DataProviderService,
+    private readonly exchangeRateDataService: ExchangeRateDataService
+  ) {}
+
+  @OnEvent(MarketDataUpdatedEvent.getName())
+  public handleMarketDataUpdated(event: MarketDataUpdatedEvent) {
+    if (
+      event.data.dataSource ===
+      this.dataProviderService.getDataSourceForExchangeRates()
+    ) {
+      this.exchangeRateDataService.invalidateCache(event.data.symbol);
+    }
+  }
+}

apps/api/src/services/exchange-rate-data/exchange-rate-data.service.ts

@@ -16,7 +16,6 @@ import {
 } from '@ghostfolio/common/helper';
 
 import { Injectable, Logger } from '@nestjs/common';
-import { OnEvent } from '@nestjs/event-emitter';
 import {
   eachDayOfInterval,
   format,
@@ -29,14 +28,15 @@ import ms from 'ms';
 
 import { ExchangeRatesByCurrency } from './interfaces/exchange-rate-data.interface';
 
+const EPOCH_OFFSET_DAYS = 25569;
+
 @Injectable()
 export class ExchangeRateDataService {
   private currencies: string[] = [];
   private currencyPairs: DataGatheringItem[] = [];
   private derivedCurrencyFactors: { [currencyPair: string]: number } = {};
   private exchangeRates: { [currencyPair: string]: number } = {};
-  private exchangeRateCache = new Map<string, Float64Array>();
+  private exchangeRateCache = new Map<string, Promise<Float64Array>>();
-  private pendingLoads = new Map<string, Promise<Float64Array>>();
 
   public constructor(
     private readonly dataProviderService: DataProviderService,
@@ -355,14 +355,12 @@ export class ExchangeRateDataService {
     return undefined;
   }
 
-  @OnEvent('market-data.updated')
+  public invalidateCache(aSymbol: string) {
-  public onMarketDataUpdated(event: { symbol: string }) {
+    this.exchangeRateCache.delete(aSymbol);
-    this.exchangeRateCache.delete(event.symbol);
-    this.pendingLoads.delete(event.symbol);
   }
 
   private getDaysSinceEpoch(aDate: Date) {
-    return Math.floor(aDate.getTime() / 86400000) + 25569;
+    return Math.floor(aDate.getTime() / ms('1d')) + EPOCH_OFFSET_DAYS;
   }
 
   private async loadCache(aSymbol: string): Promise<Float64Array> {
@@ -407,17 +405,17 @@ export class ExchangeRateDataService {
     aSymbol: string,
     aDate: Date
   ): Promise<number | undefined> {
-    let cache = this.exchangeRateCache.get(aSymbol);
+    let cachePromise = this.exchangeRateCache.get(aSymbol);
 
-    while (!cache) {
+    if (!cachePromise) {
-      if (this.pendingLoads.has(aSymbol)) {
+      cachePromise = this.loadCache(aSymbol).catch((error) => {
-        await this.pendingLoads.get(aSymbol);
+        this.exchangeRateCache.delete(aSymbol);
-        cache = this.exchangeRateCache.get(aSymbol);
+        throw error;
-      } else {
+      });
-        cache = await this.loadAndCommit(aSymbol);
+      this.exchangeRateCache.set(aSymbol, cachePromise);
-      }
     }
 
+    const cache = await cachePromise;
     const days = Math.min(this.getDaysSinceEpoch(aDate), cache.length - 1);
 
     if (days >= 0) {
@@ -428,25 +426,6 @@ export class ExchangeRateDataService {
     return undefined;
   }
 
-  private async loadAndCommit(aSymbol: string): Promise<Float64Array> {
-    const loadPromise = this.loadCache(aSymbol);
-    this.pendingLoads.set(aSymbol, loadPromise);
-
-    try {
-      const cache = await loadPromise;
-
-      if (this.pendingLoads.get(aSymbol) === loadPromise) {
-        this.exchangeRateCache.set(aSymbol, cache);
-      }
-
-      return this.exchangeRateCache.get(aSymbol) ?? cache;
-    } finally {
-      if (this.pendingLoads.get(aSymbol) === loadPromise) {
-        this.pendingLoads.delete(aSymbol);
-      }
-    }
-  }
-
   private async getExchangeRates({
     currencyFrom,
     currencyTo,

apps/api/src/services/market-data/market-data.service.ts

@@ -13,6 +13,9 @@ import {
   MarketDataState,
   Prisma
 } from '@prisma/client';
+import { uniqBy } from 'lodash';
+
+import { MarketDataUpdatedEvent } from '../../events/market-data-updated.event';
 
 @Injectable()
 export class MarketDataService {
@@ -29,7 +32,10 @@ export class MarketDataService {
       }
     });
 
-    this.eventEmitter.emit('market-data.updated', { symbol });
+    this.eventEmitter.emit(
+      MarketDataUpdatedEvent.getName(),
+      new MarketDataUpdatedEvent({ dataSource, symbol })
+    );
 
     return result;
   }
@@ -194,7 +200,10 @@ export class MarketDataService {
       }
     });
 
-    this.eventEmitter.emit('market-data.updated', { symbol });
+    this.eventEmitter.emit(
+      MarketDataUpdatedEvent.getName(),
+      new MarketDataUpdatedEvent({ dataSource, symbol })
+    );
   }
 
   public async updateAssetProfileIdentifier(
@@ -212,12 +221,14 @@ export class MarketDataService {
       }
     });
 
-    this.eventEmitter.emit('market-data.updated', {
+    this.eventEmitter.emit(
-      symbol: oldAssetProfileIdentifier.symbol
+      MarketDataUpdatedEvent.getName(),
-    });
+      new MarketDataUpdatedEvent(oldAssetProfileIdentifier)
-    this.eventEmitter.emit('market-data.updated', {
+    );
-      symbol: newAssetProfileIdentifier.symbol
+    this.eventEmitter.emit(
-    });
+      MarketDataUpdatedEvent.getName(),
+      new MarketDataUpdatedEvent(newAssetProfileIdentifier)
+    );
 
     return result;
   }
@@ -242,9 +253,13 @@ export class MarketDataService {
       update: { marketPrice: data.marketPrice, state: data.state }
     });
 
-    this.eventEmitter.emit('market-data.updated', {
+    this.eventEmitter.emit(
-      symbol: where.dataSource_date_symbol.symbol
+      MarketDataUpdatedEvent.getName(),
-    });
+      new MarketDataUpdatedEvent({
+        dataSource: where.dataSource_date_symbol.dataSource,
+        symbol: where.dataSource_date_symbol.symbol
+      })
+    );
 
     return result;
   }
@@ -285,9 +300,19 @@ export class MarketDataService {
 
     const result = await this.prismaService.$transaction(upsertPromises);
 
-    const symbols = [...new Set(data.map((d) => d.symbol as string))];
+    const uniquePairs = uniqBy(
-    for (const symbol of symbols) {
+      data.map((d) => ({
-      this.eventEmitter.emit('market-data.updated', { symbol });
+        dataSource: d.dataSource as DataSource,
+        symbol: d.symbol as string
+      })),
+      (d) => `${d.dataSource}:${d.symbol}`
+    );
+
+    for (const pair of uniquePairs) {
+      this.eventEmitter.emit(
+        MarketDataUpdatedEvent.getName(),
+        new MarketDataUpdatedEvent(pair)
+      );
     }
 
     return result;