Merge da29d2b236 into c2cb508f22

CHANGELOG.md

@@ -9,6 +9,7 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0
 
 ### Added
 
+- Added the `DATA_GATHERING_INTERVAL_MINUTES` environment variable to configure the interval of the market data gathering of the asset profiles with the hourly data gathering frequency
 - Added a link to the holdings tab in the holdings table on the overview tab of the home page (experimental)
 
 ### Changed

README.md

@@ -93,6 +93,7 @@ Find answers to commonly asked questions about self-hosting Ghostfolio in our [F
 | `API_KEY_COINGECKO_DEMO`          | `string` (optional)   |                       | The _CoinGecko_ Demo API key                                                                                                                                                                                                                                                        |
 | `API_KEY_COINGECKO_PRO`           | `string` (optional)   |                       | The _CoinGecko_ Pro API key                                                                                                                                                                                                                                                         |
 | `DATABASE_URL`                    | `string`              |                       | The database connection URL. If using a connection pooler, use the pooled connection URL here. e.g. `postgresql://${POSTGRES_USER}:${POSTGRES_PASSWORD}@localhost:5432/${POSTGRES_DB}`                                                                                              |
+| `DATA_GATHERING_INTERVAL_MINUTES` | `number` (optional)   | `60`                  | The interval in minutes of the market data gathering of the asset profiles with the hourly data gathering frequency. The data providers apply their own rate limits, so keep this value as high as your use case allows                                                             |
 | `DIRECT_URL`                      | `string` (optional)   |                       | The direct database connection URL used by the _Prisma CLI_ (e.g. for schema migrations) and seeding, bypassing any connection poolers (falls back to `DATABASE_URL`)                                                                                                               |
 | `ENABLE_FEATURE_AUTH_TOKEN`       | `boolean` (optional)  | `true`                | Enables authentication via security token                                                                                                                                                                                                                                           |
 | `ENABLE_FEATURE_MCP`              | `boolean` (optional)  | `false`               | Enables the server of the _Model Context Protocol_ (MCP) at `/mcp` (experimental)                                                                                                                                                                                                   |

apps/api/src/services/configuration/configuration.service.ts

@@ -2,6 +2,7 @@ import { environment } from '@ghostfolio/api/environments/environment';
 import { Environment } from '@ghostfolio/api/services/interfaces/environment.interface';
 import {
   CACHE_TTL_NO_CACHE,
+  DEFAULT_DATA_GATHERING_INTERVAL_MINUTES,
   DEFAULT_HOST,
   DEFAULT_PORT,
   DEFAULT_PROCESSOR_GATHER_ASSET_PROFILE_CONCURRENCY,
@@ -55,6 +56,9 @@ export class ConfigurationService {
       API_KEY_RAPID_API: str({ default: '' }),
       CACHE_QUOTES_TTL: num({ default: ms('1 minute') }),
       CACHE_TTL: num({ default: CACHE_TTL_NO_CACHE }),
+      DATA_GATHERING_INTERVAL_MINUTES: num({
+        default: DEFAULT_DATA_GATHERING_INTERVAL_MINUTES
+      }),
       DATA_SOURCE_EXCHANGE_RATES: str({ default: DataSource.YAHOO }),
       DATA_SOURCE_FEAR_AND_GREED_INDEX_STOCKS: str({
         default: DataSource.MANUAL

apps/api/src/services/cron/cron.module.ts

@@ -12,6 +12,7 @@ import { TwitterBotModule } from '@ghostfolio/api/services/twitter-bot/twitter-b
 import { TwitterBotService } from '@ghostfolio/api/services/twitter-bot/twitter-bot.service';
 
 import { Logger, Module } from '@nestjs/common';
+import { SchedulerRegistry } from '@nestjs/schedule';
 
 import { CronService } from './cron.service';
 
@@ -30,6 +31,7 @@ import { CronService } from './cron.service';
         ConfigurationService,
         DataGatheringService,
         PropertyService,
+        SchedulerRegistry,
         StatisticsGatheringService,
         TwitterBotService,
         UserService
@@ -39,6 +41,7 @@ import { CronService } from './cron.service';
         configurationService: ConfigurationService,
         dataGatheringService: DataGatheringService,
         propertyService: PropertyService,
+        schedulerRegistry: SchedulerRegistry,
         statisticsGatheringService: StatisticsGatheringService,
         twitterBotService: TwitterBotService,
         userService: UserService
@@ -53,6 +56,7 @@ import { CronService } from './cron.service';
           configurationService,
           dataGatheringService,
           propertyService,
+          schedulerRegistry,
           statisticsGatheringService,
           twitterBotService,
           userService

apps/api/src/services/cron/cron.service.spec.ts

@@ -0,0 +1,140 @@
+import { SchedulerRegistry } from '@nestjs/schedule';
+
+import { CronService } from './cron.service';
+
+describe('CronService', () => {
+  let configurationService: { get: jest.Mock };
+  let cronService: CronService;
+  let dataGatheringService: {
+    gatherHourlyMarketData: jest.Mock;
+    gatherRecentMarketData: jest.Mock;
+  };
+  let propertyService: { getByKey: jest.Mock };
+  let schedulerRegistry: SchedulerRegistry;
+
+  const createCronService = (intervalInMinutes: number) => {
+    configurationService.get.mockReturnValue(intervalInMinutes);
+
+    return new CronService(
+      configurationService as never,
+      dataGatheringService as never,
+      propertyService as never,
+      schedulerRegistry,
+      {} as never,
+      {} as never,
+      {} as never
+    );
+  };
+
+  beforeEach(() => {
+    jest.useFakeTimers();
+
+    configurationService = { get: jest.fn() };
+    dataGatheringService = {
+      gatherHourlyMarketData: jest.fn().mockResolvedValue(undefined),
+      gatherRecentMarketData: jest.fn().mockResolvedValue(undefined)
+    };
+    propertyService = { getByKey: jest.fn().mockResolvedValue(undefined) };
+    schedulerRegistry = new SchedulerRegistry();
+  });
+
+  afterEach(() => {
+    cronService?.onModuleDestroy();
+
+    jest.useRealTimers();
+  });
+
+  it('gathers the market data every hour by default', async () => {
+    cronService = createCronService(60);
+    cronService.onModuleInit();
+
+    await jest.advanceTimersByTimeAsync(59 * 60 * 1000);
+    expect(dataGatheringService.gatherHourlyMarketData).not.toHaveBeenCalled();
+
+    await jest.advanceTimersByTimeAsync(60 * 1000);
+    expect(dataGatheringService.gatherHourlyMarketData).toHaveBeenCalledTimes(
+      1
+    );
+    expect(dataGatheringService.gatherRecentMarketData).toHaveBeenCalledTimes(
+      1
+    );
+  });
+
+  it('gathers the market data in the configured interval', async () => {
+    cronService = createCronService(5);
+    cronService.onModuleInit();
+
+    await jest.advanceTimersByTimeAsync(15 * 60 * 1000);
+
+    expect(dataGatheringService.gatherHourlyMarketData).toHaveBeenCalledTimes(
+      3
+    );
+  });
+
+  it('does not gather more often than once per minute', async () => {
+    cronService = createCronService(0);
+    cronService.onModuleInit();
+
+    await jest.advanceTimersByTimeAsync(60 * 1000);
+
+    expect(dataGatheringService.gatherHourlyMarketData).toHaveBeenCalledTimes(
+      1
+    );
+  });
+
+  it('does not gather the market data if the data gathering is disabled', async () => {
+    propertyService.getByKey.mockResolvedValue(false);
+
+    cronService = createCronService(60);
+    await cronService.runHourlyMarketDataGathering();
+
+    expect(dataGatheringService.gatherHourlyMarketData).not.toHaveBeenCalled();
+  });
+
+  it('skips a run while the previous run is still in progress', async () => {
+    let finish: () => void = () => undefined;
+
+    dataGatheringService.gatherHourlyMarketData.mockReturnValue(
+      new Promise<void>((resolve) => {
+        finish = resolve;
+      })
+    );
+
+    cronService = createCronService(60);
+
+    const firstRun = cronService.runHourlyMarketDataGathering();
+
+    // Let the first run reach the pending data gathering
+    await jest.advanceTimersByTimeAsync(0);
+
+    await cronService.runHourlyMarketDataGathering();
+
+    expect(dataGatheringService.gatherHourlyMarketData).toHaveBeenCalledTimes(
+      1
+    );
+
+    finish();
+    await firstRun;
+
+    await cronService.runHourlyMarketDataGathering();
+
+    expect(dataGatheringService.gatherHourlyMarketData).toHaveBeenCalledTimes(
+      2
+    );
+  });
+
+  it('keeps running after a failed run', async () => {
+    dataGatheringService.gatherHourlyMarketData.mockRejectedValueOnce(
+      new Error('Provider unavailable')
+    );
+
+    cronService = createCronService(60);
+
+    await cronService.runHourlyMarketDataGathering();
+    await cronService.runHourlyMarketDataGathering();
+
+    expect(dataGatheringService.gatherHourlyMarketData).toHaveBeenCalledTimes(
+      2
+    );
+  });
+});

apps/api/src/services/cron/cron.service.ts

@@ -12,18 +12,28 @@ import {
 } from '@ghostfolio/common/config';
 import { getAssetProfileIdentifier } from '@ghostfolio/common/helper';
 
-import { Injectable } from '@nestjs/common';
-import { Cron, CronExpression } from '@nestjs/schedule';
+import {
+  Injectable,
+  Logger,
+  OnModuleDestroy,
+  OnModuleInit
+} from '@nestjs/common';
+import { Cron, CronExpression, SchedulerRegistry } from '@nestjs/schedule';
+import ms from 'ms';
 
 @Injectable()
-export class CronService {
-  private static readonly EVERY_HOUR_AT_RANDOM_MINUTE = `${new Date().getMinutes()} * * * *`;
+export class CronService implements OnModuleDestroy, OnModuleInit {
+  private static readonly HOURLY_MARKET_DATA_GATHERING_JOB_NAME =
+    'HOURLY_MARKET_DATA_GATHERING';
   private static readonly EVERY_SUNDAY_AT_LUNCH_TIME = '0 12 * * 0';
 
+  private isHourlyMarketDataGatheringRunning = false;
+
   public constructor(
     private readonly configurationService: ConfigurationService,
     private readonly dataGatheringService: DataGatheringService,
     private readonly propertyService: PropertyService,
+    private readonly schedulerRegistry: SchedulerRegistry,
     private readonly statisticsGatheringService: StatisticsGatheringService,
     private readonly twitterBotService: TwitterBotService,
     private readonly userService: UserService
@@ -36,11 +46,51 @@ export class CronService {
     }
   }
 
-  @Cron(CronService.EVERY_HOUR_AT_RANDOM_MINUTE)
-  public async runEveryHourAtRandomMinute() {
-    if (await this.isDataGatheringEnabled()) {
-      await this.dataGatheringService.gatherHourlyMarketData();
-      await this.dataGatheringService.gatherRecentMarketData();
+  public onModuleInit() {
+    const intervalInMinutes = Math.max(
+      1,
+      this.configurationService.get('DATA_GATHERING_INTERVAL_MINUTES')
+    );
+
+    // The interval starts at the application start, which is equivalent to the
+    // former hourly cron job running at the minute of the application start
+    this.schedulerRegistry.addInterval(
+      CronService.HOURLY_MARKET_DATA_GATHERING_JOB_NAME,
+      setInterval(
+        () => {
+          void this.runHourlyMarketDataGathering();
+        },
+        ms(`${intervalInMinutes} minutes`)
+      )
+    );
+  }
+
+  public onModuleDestroy() {
+    try {
+      this.schedulerRegistry.deleteInterval(
+        CronService.HOURLY_MARKET_DATA_GATHERING_JOB_NAME
+      );
+    } catch {}
+  }
+
+  public async runHourlyMarketDataGathering() {
+    // Skip the run if the previous one is still in progress to avoid
+    // overlapping requests to the data providers
+    if (this.isHourlyMarketDataGatheringRunning) {
+      return;
+    }
+
+    this.isHourlyMarketDataGatheringRunning = true;
+
+    try {
+      if (await this.isDataGatheringEnabled()) {
+        await this.dataGatheringService.gatherHourlyMarketData();
+        await this.dataGatheringService.gatherRecentMarketData();
+      }
+    } catch (error) {
+      Logger.error(error, 'CronService');
+    } finally {
+      this.isHourlyMarketDataGatheringRunning = false;
     }
   }
 

apps/api/src/services/interfaces/environment.interface.ts

@@ -12,6 +12,7 @@ export interface Environment extends CleanedEnvAccessors {
   API_KEY_RAPID_API: string;
   CACHE_QUOTES_TTL: number;
   CACHE_TTL: number;
+  DATA_GATHERING_INTERVAL_MINUTES: number;
   DATA_SOURCE_EXCHANGE_RATES: string;
   DATA_SOURCE_FEAR_AND_GREED_INDEX_STOCKS: string;
   DATA_SOURCE_IMPORT: string;

libs/common/src/lib/config.ts

@@ -98,6 +98,7 @@ export const PORTFOLIO_SNAPSHOT_COMPUTATION_QUEUE_PRIORITY_LOW =
 export const STATISTICS_GATHERING_QUEUE = 'STATISTICS_GATHERING_QUEUE';
 
 export const DEFAULT_CURRENCY = 'USD';
+export const DEFAULT_DATA_GATHERING_INTERVAL_MINUTES = 60;
 export const DEFAULT_DATE_FORMAT_MONTH_YEAR = 'MMM yyyy';
 export const DEFAULT_DATE_RANGE: DateRange = 'max';
 export const DEFAULT_HOST = '0.0.0.0';