ghostfolio/apps/api/src/app/endpoints/asset-profiles/asset-profiles.service.ts

import { ActivitiesService } from '@ghostfolio/api/app/activities/activities.service';
import { PortfolioChangedEvent } from '@ghostfolio/api/events/portfolio-changed.event';
import { AssetProfileSplitService } from '@ghostfolio/api/services/asset-profile-split/asset-profile-split.service';
import { BenchmarkService } from '@ghostfolio/api/services/benchmark/benchmark.service';
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 { MarketDataService } from '@ghostfolio/api/services/market-data/market-data.service';
import { PrismaService } from '@ghostfolio/api/services/prisma/prisma.service';
import { DataGatheringService } from '@ghostfolio/api/services/queues/data-gathering/data-gathering.service';
import { SymbolProfileService } from '@ghostfolio/api/services/symbol-profile/symbol-profile.service';
import { UpdateAssetProfileDataDto } from '@ghostfolio/common/dtos';
import {
  applyAssetProfileOverrides,
  getAssetProfileIdentifier,
  getCurrencyFromSymbol,
  isCurrency
} from '@ghostfolio/common/helper';
import {
  AdminMarketDataDetails,
  AssetProfileIdentifier,
  AssetProfileItem,
  AssetProfilesResponse,
  DataProviderInfo,
  EnhancedAssetProfile,
  Filter
} from '@ghostfolio/common/interfaces';
import { MarketDataPreset } from '@ghostfolio/common/types';

import { Injectable, Logger, NotFoundException } from '@nestjs/common';
import { EventEmitter2 } from '@nestjs/event-emitter';
import { AssetClass, AssetSubClass, DataSource, Prisma } from '@prisma/client';
import { groupBy } from 'lodash-es';

@Injectable()
export class AssetProfilesService {
  private readonly logger = new Logger(AssetProfilesService.name);

  public constructor(
    private readonly activitiesService: ActivitiesService,
    private readonly assetProfileSplitService: AssetProfileSplitService,
    private readonly benchmarkService: BenchmarkService,
    private readonly dataGatheringService: DataGatheringService,
    private readonly dataProviderService: DataProviderService,
    private readonly eventEmitter: EventEmitter2,
    private readonly exchangeRateDataService: ExchangeRateDataService,
    private readonly marketDataService: MarketDataService,
    private readonly prismaService: PrismaService,
    private readonly symbolProfileService: SymbolProfileService
  ) {}

  public async createSplit({
    dataSource,
    date,
    denominator,
    numerator,
    symbol,
    symbolProfileId
  }: {
    date: Date;
    denominator: number;
    numerator: number;
    symbolProfileId: string;
  } & AssetProfileIdentifier) {
    const assetProfileSplit = await this.assetProfileSplitService.upsert({
      date,
      denominator,
      numerator,
      symbolProfileId
    });

    await this.gatherSymbolAndEmitPortfolioChangedEvents({
      dataSource,
      symbol,
      symbolProfileId
    });

    return assetProfileSplit;
  }

  public async deleteSplit({
    dataSource,
    id,
    symbol,
    symbolProfileId
  }: {
    id: string;
    symbolProfileId: string;
  } & AssetProfileIdentifier) {
    const isDeleted = await this.assetProfileSplitService.deleteById({
      id,
      symbolProfileId
    });

    if (!isDeleted) {
      throw new NotFoundException();
    }

    await this.gatherSymbolAndEmitPortfolioChangedEvents({
      dataSource,
      symbol,
      symbolProfileId
    });
  }

  /**
   * Gathers the market data of the given asset profile and invalidates the
   * portfolio snapshots of the affected users as soon as it is available.
   * Emitting the events earlier would recompute the snapshots from
   * split-adjusted quantities and not yet split-adjusted market prices.
   *
   * With withImmediateInvalidation the snapshots are invalidated a second
   * time, before the gathering. This is necessary if the snapshots are already
   * wrong without new market data, because the invalidation after the
   * gathering is held in memory and is therefore lost if the process restarts.
   */
  public async gatherSymbolAndEmitPortfolioChangedEvents({
    dataSource,
    force = true,
    symbol,
    symbolProfileId,
    withImmediateInvalidation = false
  }: {
    force?: boolean;
    symbolProfileId: string;
    withImmediateInvalidation?: boolean;
  } & AssetProfileIdentifier) {
    if (withImmediateInvalidation) {
      await this.emitPortfolioChangedEvents(symbolProfileId);
    }

    const jobs = await this.dataGatheringService.gatherSymbol({
      dataSource,
      force,
      symbol
    });

    void Promise.allSettled(
      jobs.map((job) => {
        return job.finished();
      })
    )
      .then(() => {
        return this.emitPortfolioChangedEvents(symbolProfileId);
      })
      .catch((error) => {
        this.logger.error(
          `Could not emit the portfolio changed events of the asset profile ${getAssetProfileIdentifier({ dataSource, symbol })}`,
          error.stack
        );
      });
  }

  public async getAssetProfile({
    dataSource,
    symbol
  }: AssetProfileIdentifier): Promise<AdminMarketDataDetails> {
    let activitiesCount: EnhancedAssetProfile['activitiesCount'] = 0;
    let currency: EnhancedAssetProfile['currency'] = '-';
    let dateOfFirstActivity: EnhancedAssetProfile['dateOfFirstActivity'];

    const isCurrencyAssetProfile = isCurrency(getCurrencyFromSymbol(symbol));

    if (isCurrencyAssetProfile) {
      currency = getCurrencyFromSymbol(symbol);
      ({ activitiesCount, dateOfFirstActivity } =
        await this.activitiesService.getStatisticsByCurrency(currency));
    }

    const [[assetProfile], marketData, splits] = await Promise.all([
      this.symbolProfileService.getSymbolProfiles([
        {
          dataSource,
          symbol
        }
      ]),
      this.marketDataService.marketDataItems({
        orderBy: {
          date: 'asc'
        },
        where: {
          dataSource,
          symbol
        }
      }),
      this.assetProfileSplitService.getSplits({ dataSource, symbol })
    ]);

    if (assetProfile) {
      assetProfile.dataProviderInfo = this.dataProviderService
        .getDataProvider(assetProfile.dataSource)
        .getDataProviderInfo();
    }

    return {
      marketData,
      splits,
      assetProfile: assetProfile ?? {
        activitiesCount,
        currency,
        dataSource,
        dateOfFirstActivity,
        symbol,
        assetClass: isCurrencyAssetProfile ? AssetClass.LIQUIDITY : undefined,
        assetSubClass: isCurrencyAssetProfile ? AssetSubClass.CASH : undefined,
        isActive: true
      }
    };
  }

  public async getAssetProfiles({
    filters = [],
    presetId,
    sortColumn,
    sortDirection = 'asc',
    skip,
    take = Number.MAX_SAFE_INTEGER
  }: {
    filters?: Filter[];
    presetId?: MarketDataPreset;
    skip?: number;
    sortColumn?: string;
    sortDirection?: Prisma.SortOrder;
    take?: number;
  }): Promise<AssetProfilesResponse> {
    let orderBy: Prisma.Enumerable<Prisma.SymbolProfileOrderByWithRelationInput> =
      [{ symbol: 'asc' }];
    const where: Prisma.SymbolProfileWhereInput = {};

    if (presetId === 'BENCHMARKS') {
      const benchmarkAssetProfiles =
        await this.benchmarkService.getBenchmarkAssetProfiles();

      where.id = {
        in: benchmarkAssetProfiles.map(({ id }) => {
          return id;
        })
      };
    } else if (presetId === 'CURRENCIES') {
      return this.getAssetProfilesForCurrencies();
    } else if (
      presetId === 'ETF_WITHOUT_COUNTRIES' ||
      presetId === 'ETF_WITHOUT_SECTORS'
    ) {
      filters = [{ id: 'ETF', type: 'ASSET_SUB_CLASS' }];
    } else if (presetId === 'NO_ACTIVITIES') {
      where.activities = {
        none: {}
      };
    }

    const {
      ASSET_SUB_CLASS: [filterByAssetSubClass] = [],
      DATA_SOURCE: [filterByDataSource] = [],
      SEARCH_QUERY: [filterBySearchQuery] = []
    } = groupBy(filters, ({ type }) => {
      return type;
    });

    const marketDataItems = await this.prismaService.marketData.groupBy({
      _count: true,
      by: ['dataSource', 'symbol']
    });

    if (filterByAssetSubClass) {
      where.assetSubClass = AssetSubClass[filterByAssetSubClass.id];
    }

    if (filterByDataSource) {
      where.dataSource = DataSource[filterByDataSource.id];
    }

    if (filterBySearchQuery) {
      where.OR = [
        { id: { mode: 'insensitive', startsWith: filterBySearchQuery.id } },
        { isin: { mode: 'insensitive', startsWith: filterBySearchQuery.id } },
        { name: { mode: 'insensitive', startsWith: filterBySearchQuery.id } },
        { symbol: { mode: 'insensitive', startsWith: filterBySearchQuery.id } }
      ];
    }

    if (sortColumn) {
      orderBy = [{ [sortColumn]: sortDirection }];

      if (sortColumn === 'activitiesCount') {
        orderBy = [
          {
            activities: {
              _count: sortDirection
            }
          }
        ];
      }
    }

    const extendedPrismaClient = this.getExtendedPrismaClient();

    const symbolProfileResult = await Promise.all([
      extendedPrismaClient.symbolProfile.findMany({
        skip,
        take,
        where,
        orderBy: [...orderBy, { id: sortDirection }],
        select: {
          _count: {
            select: {
              activities: true,
              watchedBy: true
            }
          },
          activities: {
            orderBy: [{ date: 'asc' }],
            select: { date: true },
            take: 1
          },
          assetClass: true,
          assetProfileOverrides: true,
          assetSubClass: true,
          comment: true,
          countries: true,
          currency: true,
          dataSource: true,
          id: true,
          isin: true,
          isActive: true,
          isUsedByUsersWithSubscription: true,
          name: true,
          scraperConfiguration: true,
          sectors: true,
          symbol: true
        }
      }),
      this.prismaService.symbolProfile.count({ where })
    ]);
    const symbolProfiles = symbolProfileResult[0];
    let count = symbolProfileResult[1];

    const lastMarketPrices = await this.prismaService.marketData.findMany({
      distinct: ['dataSource', 'symbol'],
      orderBy: { date: 'desc' },
      select: {
        dataSource: true,
        marketPrice: true,
        symbol: true
      },
      where: {
        dataSource: {
          in: symbolProfiles.map(({ dataSource }) => {
            return dataSource;
          })
        },
        symbol: {
          in: symbolProfiles.map(({ symbol }) => {
            return symbol;
          })
        }
      }
    });

    const lastMarketPriceMap = new Map<string, number>();

    for (const { dataSource, marketPrice, symbol } of lastMarketPrices) {
      lastMarketPriceMap.set(
        getAssetProfileIdentifier({ dataSource, symbol }),
        marketPrice
      );
    }

    const dataProviderInfoMap = new Map<DataSource, DataProviderInfo>(
      [
        ...new Set(
          symbolProfiles.map(({ dataSource }) => {
            return dataSource;
          })
        )
      ].map((dataSource) => {
        return [
          dataSource,
          this.dataProviderService
            .getDataProvider(dataSource)
            .getDataProviderInfo()
        ];
      })
    );

    let assetProfiles: AssetProfileItem[] = await Promise.all(
      symbolProfiles.map(async (assetProfile) => {
        const {
          _count,
          activities,
          comment,
          currency,
          dataSource,
          id,
          isin,
          isActive,
          isUsedByUsersWithSubscription,
          symbol
        } = assetProfile;

        const { assetClass, assetSubClass, countries, name, sectors } =
          applyAssetProfileOverrides(
            assetProfile,
            assetProfile.assetProfileOverrides
          );

        const countriesCount = countries ? Object.keys(countries).length : 0;

        const lastMarketPrice = lastMarketPriceMap.get(
          getAssetProfileIdentifier({ dataSource, symbol })
        );

        const marketDataItemCount =
          marketDataItems.find((marketDataItem) => {
            return (
              marketDataItem.dataSource === dataSource &&
              marketDataItem.symbol === symbol
            );
          })?._count ?? 0;

        const sectorsCount = sectors ? Object.keys(sectors).length : 0;

        return {
          assetClass,
          assetSubClass,
          comment,
          countriesCount,
          currency,
          dataSource,
          id,
          isActive,
          isin,
          lastMarketPrice,
          marketDataItemCount,
          name,
          sectorsCount,
          symbol,
          activitiesCount: _count.activities,
          dataProviderInfo: dataProviderInfoMap.get(dataSource),
          date: activities?.[0]?.date,
          isUsedByUsersWithSubscription: await isUsedByUsersWithSubscription,
          watchedByCount: _count.watchedBy
        };
      })
    );

    if (presetId) {
      if (presetId === 'ETF_WITHOUT_COUNTRIES') {
        assetProfiles = assetProfiles.filter(({ countriesCount }) => {
          return countriesCount === 0;
        });
      } else if (presetId === 'ETF_WITHOUT_SECTORS') {
        assetProfiles = assetProfiles.filter(({ sectorsCount }) => {
          return sectorsCount === 0;
        });
      }

      count = assetProfiles.length;
    }

    return {
      assetProfiles,
      count
    };
  }

  public async updateAssetProfileData(
    { dataSource, symbol }: AssetProfileIdentifier,
    assetProfileData: UpdateAssetProfileDataDto
  ): Promise<EnhancedAssetProfile> {
    const notFoundMessage = `Could not find the asset profile for ${symbol} (${dataSource})`;

    const data = this.getAssetProfileDataUpdate(assetProfileData);

    if (Object.keys(data).length > 0) {
      try {
        await this.symbolProfileService.updateSymbolProfile(
          {
            dataSource,
            symbol
          },
          this.symbolProfileService.getAssetProfileUpdateInput(
            { dataSource, symbol },
            data
          )
        );
      } catch (error) {
        if (
          error instanceof Prisma.PrismaClientKnownRequestError &&
          error.code === 'P2025'
        ) {
          throw new NotFoundException(notFoundMessage);
        }

        throw error;
      }
    }

    const [assetProfile] = await this.symbolProfileService.getSymbolProfiles([
      {
        dataSource,
        symbol
      }
    ]);

    if (!assetProfile) {
      throw new NotFoundException(notFoundMessage);
    }

    return assetProfile;
  }

  private async emitPortfolioChangedEvents(symbolProfileId: string) {
    const userIds =
      await this.activitiesService.getUserIdsBySymbolProfileId(symbolProfileId);

    for (const userId of userIds) {
      this.eventEmitter.emit(
        PortfolioChangedEvent.getName(),
        new PortfolioChangedEvent({ userId })
      );
    }
  }

  private getAssetProfileDataUpdate({
    countries,
    holdings,
    sectors
  }: UpdateAssetProfileDataDto): Pick<
    Prisma.SymbolProfileUpdateInput,
    'countries' | 'holdings' | 'sectors'
  > {
    const data: Pick<
      Prisma.SymbolProfileUpdateInput,
      'countries' | 'holdings' | 'sectors'
    > = {};

    if (countries !== undefined) {
      data.countries = countries as Prisma.JsonArray;
    }

    if (holdings !== undefined) {
      data.holdings = holdings as Prisma.JsonArray;
    }

    if (sectors !== undefined) {
      data.sectors = sectors as Prisma.JsonArray;
    }

    return data;
  }

  private async getAssetProfilesForCurrencies(): Promise<AssetProfilesResponse> {
    const currencyPairs = this.exchangeRateDataService.getCurrencyPairs();

    const [lastMarketPrices, marketDataItems] = await Promise.all([
      this.prismaService.marketData.findMany({
        distinct: ['dataSource', 'symbol'],
        orderBy: { date: 'desc' },
        select: {
          dataSource: true,
          marketPrice: true,
          symbol: true
        },
        where: {
          dataSource: {
            in: currencyPairs.map(({ dataSource }) => {
              return dataSource;
            })
          },
          symbol: {
            in: currencyPairs.map(({ symbol }) => {
              return symbol;
            })
          }
        }
      }),
      this.prismaService.marketData.groupBy({
        _count: true,
        by: ['dataSource', 'symbol']
      })
    ]);

    const lastMarketPriceMap = new Map<string, number>();

    for (const { dataSource, marketPrice, symbol } of lastMarketPrices) {
      lastMarketPriceMap.set(
        getAssetProfileIdentifier({ dataSource, symbol }),
        marketPrice
      );
    }

    const assetProfilePromises: Promise<AssetProfileItem>[] = currencyPairs.map(
      async ({ dataSource, symbol }) => {
        let activitiesCount: EnhancedAssetProfile['activitiesCount'] = 0;
        let currency: EnhancedAssetProfile['currency'] = '-';
        let dateOfFirstActivity: EnhancedAssetProfile['dateOfFirstActivity'];

        if (isCurrency(getCurrencyFromSymbol(symbol))) {
          currency = getCurrencyFromSymbol(symbol);
          ({ activitiesCount, dateOfFirstActivity } =
            await this.activitiesService.getStatisticsByCurrency(currency));
        }

        const lastMarketPrice = lastMarketPriceMap.get(
          getAssetProfileIdentifier({ dataSource, symbol })
        );

        const marketDataItemCount =
          marketDataItems.find((marketDataItem) => {
            return (
              marketDataItem.dataSource === dataSource &&
              marketDataItem.symbol === symbol
            );
          })?._count ?? 0;

        return {
          activitiesCount,
          currency,
          dataSource,
          lastMarketPrice,
          marketDataItemCount,
          symbol,
          assetClass: AssetClass.LIQUIDITY,
          assetSubClass: AssetSubClass.CASH,
          countriesCount: 0,
          date: dateOfFirstActivity,
          id: undefined,
          isActive: true,
          name: symbol,
          sectorsCount: 0,
          watchedByCount: 0
        };
      }
    );

    const assetProfiles = await Promise.all(assetProfilePromises);
    return { assetProfiles, count: assetProfiles.length };
  }

  private getExtendedPrismaClient() {
    const symbolProfileExtension = Prisma.defineExtension((client) => {
      return client.$extends({
        result: {
          symbolProfile: {
            isUsedByUsersWithSubscription: {
              compute: async ({ id }) => {
                const { _count } =
                  await this.prismaService.symbolProfile.findUnique({
                    select: {
                      _count: {
                        select: {
                          activities: {
                            where: {
                              user: {
                                subscriptions: {
                                  some: {
                                    expiresAt: {
                                      gt: new Date()
                                    }
                                  }
                                }
                              }
                            }
                          }
                        }
                      }
                    },
                    where: {
                      id
                    }
                  });

                return _count.activities > 0;
              }
            }
          }
        }
      });
    });

    return this.prismaService.$extends(symbolProfileExtension);
  }
}