From 15857118febdc1316aff87b38badc549ccdb2b4d Mon Sep 17 00:00:00 2001
From: Thomas Kaul <4159106+dtslvr@users.noreply.github.com>
Date: Wed, 17 Apr 2024 17:35:51 +0200
Subject: [PATCH] Feature/let data gathering queue jobs fail by throwing errors
 (#3281)

* Let data gathering queue jobs fail by throwing errors

* Update changelog
---
 CHANGELOG.md                                  |  2 ++
 .../src/app/portfolio/portfolio.service.ts    | 18 +++++++++++----
 apps/api/src/app/symbol/symbol.service.ts     | 20 ++++++++++++----
 .../data-gathering.processor.ts               | 20 ++++++++++++----
 .../data-gathering/data-gathering.service.ts  | 23 ++++++++-----------
 .../data-provider/data-provider.service.ts    | 18 +++++++++------
 6 files changed, 65 insertions(+), 36 deletions(-)

diff --git a/CHANGELOG.md b/CHANGELOG.md
index 8e26c136..2941ab9e 100644
--- a/CHANGELOG.md
+++ b/CHANGELOG.md
@@ -19,6 +19,8 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0
 - Moved the interest calculations into the portfolio calculator
 - Moved the liability calculations into the portfolio calculator
 - Moved the (wealth) item calculations into the portfolio calculator
+- Let queue jobs for asset profile data gathering fail by throwing an error
+- Let queue jobs for historical market data gathering fail by throwing an error
 
 ## 2.72.0 - 2024-04-13
 
diff --git a/apps/api/src/app/portfolio/portfolio.service.ts b/apps/api/src/app/portfolio/portfolio.service.ts
index 95a68eaa..4c65d2eb 100644
--- a/apps/api/src/app/portfolio/portfolio.service.ts
+++ b/apps/api/src/app/portfolio/portfolio.service.ts
@@ -822,11 +822,19 @@ export class PortfolioService {
       );
 
       if (isEmpty(historicalData)) {
-        historicalData = await this.dataProviderService.getHistoricalRaw(
-          [{ dataSource: DataSource.YAHOO, symbol: aSymbol }],
-          portfolioStart,
-          new Date()
-        );
+        try {
+          historicalData = await this.dataProviderService.getHistoricalRaw({
+            dataGatheringItems: [
+              { dataSource: DataSource.YAHOO, symbol: aSymbol }
+            ],
+            from: portfolioStart,
+            to: new Date()
+          });
+        } catch {
+          historicalData = {
+            [aSymbol]: {}
+          };
+        }
       }
 
       const historicalDataArray: HistoricalDataItem[] = [];
diff --git a/apps/api/src/app/symbol/symbol.service.ts b/apps/api/src/app/symbol/symbol.service.ts
index 9a3f7a3a..90259a77 100644
--- a/apps/api/src/app/symbol/symbol.service.ts
+++ b/apps/api/src/app/symbol/symbol.service.ts
@@ -74,11 +74,21 @@ export class SymbolService {
     date = new Date(),
     symbol
   }: IDataGatheringItem): Promise<IDataProviderHistoricalResponse> {
-    const historicalData = await this.dataProviderService.getHistoricalRaw(
-      [{ dataSource, symbol }],
-      date,
-      date
-    );
+    let historicalData: {
+      [symbol: string]: {
+        [date: string]: IDataProviderHistoricalResponse;
+      };
+    } = {
+      [symbol]: {}
+    };
+
+    try {
+      historicalData = await this.dataProviderService.getHistoricalRaw({
+        dataGatheringItems: [{ dataSource, symbol }],
+        from: date,
+        to: date
+      });
+    } catch {}
 
     return {
       marketPrice:
diff --git a/apps/api/src/services/data-gathering/data-gathering.processor.ts b/apps/api/src/services/data-gathering/data-gathering.processor.ts
index bf960048..11eda2e7 100644
--- a/apps/api/src/services/data-gathering/data-gathering.processor.ts
+++ b/apps/api/src/services/data-gathering/data-gathering.processor.ts
@@ -37,7 +37,17 @@ export class DataGatheringProcessor {
   @Process({ concurrency: 1, name: GATHER_ASSET_PROFILE_PROCESS })
   public async gatherAssetProfile(job: Job<UniqueAsset>) {
     try {
+      Logger.log(
+        `Asset profile data gathering has been started for ${job.data.symbol} (${job.data.dataSource})`,
+        `DataGatheringProcessor (${GATHER_ASSET_PROFILE_PROCESS})`
+      );
+
       await this.dataGatheringService.gatherAssetProfiles([job.data]);
+
+      Logger.log(
+        `Asset profile data gathering has been completed for ${job.data.symbol} (${job.data.dataSource})`,
+        `DataGatheringProcessor (${GATHER_ASSET_PROFILE_PROCESS})`
+      );
     } catch (error) {
       Logger.error(
         error,
@@ -62,11 +72,11 @@ export class DataGatheringProcessor {
         `DataGatheringProcessor (${GATHER_HISTORICAL_MARKET_DATA_PROCESS})`
       );
 
-      const historicalData = await this.dataProviderService.getHistoricalRaw(
-        [{ dataSource, symbol }],
-        currentDate,
-        new Date()
-      );
+      const historicalData = await this.dataProviderService.getHistoricalRaw({
+        dataGatheringItems: [{ dataSource, symbol }],
+        from: currentDate,
+        to: new Date()
+      });
 
       const data: Prisma.MarketDataUpdateInput[] = [];
       let lastMarketPrice: number;
diff --git a/apps/api/src/services/data-gathering/data-gathering.service.ts b/apps/api/src/services/data-gathering/data-gathering.service.ts
index b2b0c371..31a0040a 100644
--- a/apps/api/src/services/data-gathering/data-gathering.service.ts
+++ b/apps/api/src/services/data-gathering/data-gathering.service.ts
@@ -104,11 +104,11 @@ export class DataGatheringService {
     symbol: string;
   }) {
     try {
-      const historicalData = await this.dataProviderService.getHistoricalRaw(
-        [{ dataSource, symbol }],
-        date,
-        date
-      );
+      const historicalData = await this.dataProviderService.getHistoricalRaw({
+        dataGatheringItems: [{ dataSource, symbol }],
+        from: date,
+        to: date
+      });
 
       const marketPrice =
         historicalData[symbol][format(date, DATE_FORMAT)].marketPrice;
@@ -230,17 +230,12 @@ export class DataGatheringService {
           error,
           'DataGatheringService'
         );
+
+        if (uniqueAssets.length === 1) {
+          throw error;
+        }
       }
     }
-
-    Logger.log(
-      `Asset profile data gathering has been completed for ${uniqueAssets
-        .map(({ dataSource, symbol }) => {
-          return `${symbol} (${dataSource})`;
-        })
-        .join(',')}.`,
-      'DataGatheringService'
-    );
   }
 
   public async gatherSymbols({
diff --git a/apps/api/src/services/data-provider/data-provider.service.ts b/apps/api/src/services/data-provider/data-provider.service.ts
index 9afdab1d..675c1337 100644
--- a/apps/api/src/services/data-provider/data-provider.service.ts
+++ b/apps/api/src/services/data-provider/data-provider.service.ts
@@ -233,15 +233,17 @@ export class DataProviderService {
     }
   }
 
-  public async getHistoricalRaw(
-    aDataGatheringItems: UniqueAsset[],
-    from: Date,
-    to: Date
-  ): Promise<{
+  public async getHistoricalRaw({
+    dataGatheringItems,
+    from,
+    to
+  }: {
+    dataGatheringItems: UniqueAsset[];
+    from: Date;
+    to: Date;
+  }): Promise<{
     [symbol: string]: { [date: string]: IDataProviderHistoricalResponse };
   }> {
-    let dataGatheringItems = aDataGatheringItems;
-
     for (const { currency, rootCurrency } of DERIVED_CURRENCIES) {
       if (
         this.hasCurrency({
@@ -330,6 +332,8 @@ export class DataProviderService {
       }
     } catch (error) {
       Logger.error(error, 'DataProviderService');
+
+      throw error;
     }
 
     return result;