Merge pull request #43 from dandevaud/bugfix/several-bugfixes

Readded Locks

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

@@ -24,6 +24,7 @@ import { DataSource } from '@prisma/client';
 import { JobOptions, Queue } from 'bull';
 import { format, min, subDays, subYears } from 'date-fns';
 import { isEmpty } from 'lodash';
+import AwaitLock from 'await-lock';
 
 @Injectable()
 export class DataGatheringService {
@@ -40,6 +41,8 @@ export class DataGatheringService {
     private readonly symbolProfileService: SymbolProfileService
   ) {}
 
+  lock = new AwaitLock();
+
   public async addJobToQueue({
     data,
     name,
@@ -100,6 +103,8 @@ export class DataGatheringService {
         historicalData[symbol][format(date, DATE_FORMAT)].marketPrice;
 
       if (marketPrice) {
+        await this.lock.acquireAsync();
+        try {
           return await this.prismaService.marketData.upsert({
             create: {
               dataSource,
@@ -110,6 +115,9 @@ export class DataGatheringService {
             update: { marketPrice },
             where: { dataSource_date_symbol: { dataSource, date, symbol } }
           });
+        } finally {
+          this.lock.release();
+        }
       }
     } catch (error) {
       Logger.error(error, 'DataGatheringService');

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

@@ -13,11 +13,14 @@ import {
   Prisma
 } from '@prisma/client';
 import { DateQueryHelper } from '@ghostfolio/api/helper/dateQueryHelper';
+import AwaitLock from 'await-lock';
 
 @Injectable()
 export class MarketDataService {
   public constructor(private readonly prismaService: PrismaService) {}
 
+  lock = new AwaitLock();
+
   private dateQueryHelper = new DateQueryHelper();
 
   public async deleteMany({ dataSource, symbol }: UniqueAsset) {
@@ -129,6 +132,8 @@ export class MarketDataService {
     where: Prisma.MarketDataWhereUniqueInput;
   }): Promise<MarketData> {
     const { data, where } = params;
+    await this.lock.acquireAsync();
+    try {
       return this.prismaService.marketData.upsert({
         where,
         create: {
@@ -140,6 +145,9 @@ export class MarketDataService {
         },
         update: { marketPrice: data.marketPrice, state: data.state }
       });
+    } finally {
+      this.lock.release();
+    }
   }
 
   /**
@@ -152,7 +160,9 @@ export class MarketDataService {
     data: Prisma.MarketDataUpdateInput[];
   }): Promise<MarketData[]> {
     const upsertPromises = data.map(
-      ({ dataSource, date, marketPrice, symbol, state }) => {
+      async ({ dataSource, date, marketPrice, symbol, state }) => {
+        await this.lock.acquireAsync();
+        try {
           return this.prismaService.marketData.upsert({
             create: {
               dataSource: <DataSource>dataSource,
@@ -173,9 +183,11 @@ export class MarketDataService {
               }
             }
           });
+        } finally {
+          this.lock.release();
+        }
       }
     );
-
-    return this.prismaService.$transaction(upsertPromises);
+    return await Promise.all(upsertPromises);
   }
 }

package.json

@@ -86,6 +86,7 @@
     "@simplewebauthn/server": "8.3.2",
     "@stripe/stripe-js": "1.47.0",
     "alphavantage": "2.2.0",
+    "await-lock": "^2.2.2",
     "big.js": "6.2.1",
     "body-parser": "1.20.1",
     "bootstrap": "4.6.0",

yarn.lock

@@ -7366,6 +7366,11 @@ available-typed-arrays@^1.0.5:
   resolved "https://registry.yarnpkg.com/available-typed-arrays/-/available-typed-arrays-1.0.5.tgz#92f95616501069d07d10edb2fc37d3e1c65123b7"
   integrity sha512-DMD0KiN46eipeziST1LPP/STfDU0sufISXmjSgvVsoU2tqxctQeASejWcfNtxYKqETM1UxQ8sp2OrSBWpHY6sw==
 
+await-lock@^2.2.2:
+  version "2.2.2"
+  resolved "https://registry.yarnpkg.com/await-lock/-/await-lock-2.2.2.tgz#a95a9b269bfd2f69d22b17a321686f551152bcef"
+  integrity sha512-aDczADvlvTGajTDjcjpJMqRkOF6Qdz3YbPZm/PyW6tKPkx2hlYBzxMhEywM/tU72HrVZjgl5VCdRuMlA7pZ8Gw==
+
 aws-sign2@~0.7.0:
   version "0.7.0"
   resolved "https://registry.yarnpkg.com/aws-sign2/-/aws-sign2-0.7.0.tgz#b46e890934a9591f2d2f6f86d7e6a9f1b3fe76a8"