backend/src/modules/investments/portfolio/application/portfolio.service.ts

import {
  Injectable,
  NotFoundException,
  ServiceUnavailableException,
} from '@nestjs/common';
import { randomUUID } from 'node:crypto';

import {
  add,
  decimalFromString,
  formatDecimal,
  isZero,
  multiply,
  subtract,
} from '../../../../common/decimal';
import { RequestContextService } from '../../../../common/logging/request-context.service';
import { StructuredLogger } from '../../../../common/logging/structured-logger.service';
import { MoneyDto } from '../../../../common/money.dto';
import { decimal } from '../../../../integrations/tbank/decimal';
import { isTbankErrorCode, TbankErrorCode } from '../../../../integrations/tbank/tbank-error-code';
import {
  InstrumentMetadataService,
  type MetadataIdentity,
  metadataKey,
  TbankAccessService,
  TbankClient,
  type TbankInstrument,
  TbankMembershipService,
  TbankMoney,
  tbankSourceTtlMs,
} from '../../../../integrations/tbank/tbank.api';
import { CandleService } from '../../instruments/instruments.api';
import { investmentResourceMeta } from '../../investment-resource.mapper';
import { portfolioStructure } from '../domain/portfolio-structure';
import { sumMoneyByCurrency } from '../domain/portfolio-totals';
import {
  portfolioFreshForMs,
  PortfolioRepository,
  portfolioWorkKey,
} from '../infrastructure/portfolio.repository';
import { totalsByCurrency } from '../infrastructure/tbank-portfolio.mapper';
import {
  PortfolioDto,
  PortfolioPositionDto,
} from '../presentation/portfolio.dto';

const MISSING_DATA_REVISION = 0;

const membershipFreshForMs = tbankSourceTtlMs.membership;
const types: Record<string, string> = {
  share: 'STOCK',
  bond: 'BOND',
  etf: 'FUND',
  currency: 'CURRENCY',
  futures: 'FUTURE',
  option: 'OPTION',
  sp: 'STRUCTURED_NOTE',
  dfa: 'DIGITAL_FINANCIAL_ASSET',
};

@Injectable()
export class PortfolioService {
  private readonly processRunId = randomUUID();
  private readonly active = new Set<string>();
  private readonly pending = new Map<string, Promise<void>>();
  private reservations: Promise<unknown> = Promise.resolve();

  constructor(
    private readonly access: TbankAccessService,
    private readonly tbank: TbankClient,
    private readonly requestContext: RequestContextService,
    private readonly metadata: InstrumentMetadataService,
    private readonly repository: PortfolioRepository,
    private readonly membership: TbankMembershipService,
    private readonly logger: StructuredLogger,
    private readonly candles: CandleService,
  ) {}

  async saved(userId: string, accountId: string, observe = false) {
    this.requestContext.set({ userId, brokerageAccountId: accountId });
    const connection = await this.access.local(userId, accountId);
    if (!observe) {
      const discovery = await this.membership.request(
        userId,
        connection.revision,
      );
      if (discovery) {
        void discovery.done
          .then(async () => {
            const current = await this.access.local(userId, accountId);
            if (current.revision !== connection.revision) return;
            for (const account of current.accounts.filter(
              (account) =>
                accountId === 'all' || account.accountId === accountId,
            )) {
              await this.queueSnapshot(
                userId,
                current.revision,
                account.accountId,
              );
            }
          })
          .catch(() =>
            this.logger.event(
              'warn',
              'tbank.portfolio.discovery_not_available',
              { userId },
            ),
          );
      }
      for (const account of (discovery ? [] : connection.accounts).filter(
        (account) => accountId === 'all' || account.accountId === accountId,
      )) {
        await this.queueSnapshot(
          userId,
          connection.revision,
          account.accountId,
        );
      }
    }
    const saved = await this.repository.read(userId, connection.revision);
    if (
      saved?.membershipAt &&
      accountId !== 'all' &&
      !saved.accounts.some((account) => account.accountId === accountId)
    ) {
      throw new NotFoundException();
    }
    if (!observe) {
      const identities = new Map<string, MetadataIdentity>();
      for (const account of (saved?.accounts ?? []).filter(
        (account) => accountId === 'all' || account.accountId === accountId,
      )) {
        const snapshot = account.portfolio?.payload as unknown as
          PortfolioDto | undefined;
        for (const position of snapshot?.positions ?? []) {
          identities.set(metadataKey(position), position);
        }
      }
      const savedMetadata = await this.metadata.savedMany([
        ...identities.values(),
      ]);
      for (const identity of identities.values()) {
        if (!savedMetadata.get(metadataKey(identity))?.fresh) {
          await this.metadata.request(userId, connection.revision, identity);
        }
      }
    }
    if (!observe) {
      void this.candles
        .prefetch(
          userId,
          connection.revision,
          (saved?.accounts ?? [])
            .filter(
              (account) =>
                accountId === 'all' || account.accountId === accountId,
            )
            .flatMap(
              (account) =>
                (
                  account.portfolio?.payload as unknown as
                    PortfolioDto | undefined
                )?.positions ?? [],
            ),
        )
        .catch(() =>
          this.logger.event('warn', 'portfolio.candle-prefetch-unavailable'),
        );
    }
    const membershipMeta = investmentResourceMeta({
      resourceKey: 'accounts/membership',
      dataRevision: String(saved?.membershipRevision ?? MISSING_DATA_REVISION),
      availability: saved?.membershipAt ? 'complete' : 'missing',
      fetchedAt: saved?.membershipAt ?? null,
      freshForMs: membershipFreshForMs,
      work: saved?.work.find((work) => work.scopeKey === 'membership') ?? null,
    });
    return {
      membership: membershipMeta,
      accounts: (saved?.accounts ?? [])
        .filter(
          (account) => accountId === 'all' || account.accountId === accountId,
        )
        .map((account) => ({
          id: account.accountId,
          name: account.name,
          type: account.type,
          openedOn: account.openedOn,
          snapshot: account.portfolio
            ? {
                ...(account.portfolio.payload as unknown as PortfolioDto),
                lastSuccessfulAt: account.portfolio.fetchedAt.toISOString(),
                isStale:
                  Date.now() - account.portfolio.fetchedAt.valueOf() >=
                  portfolioFreshForMs,
              }
            : null,
          meta: investmentResourceMeta({
            resourceKey: `portfolio/${encodeURIComponent(account.accountId)}`,
            dataRevision: String(account.portfolio?.dataRevision ?? MISSING_DATA_REVISION),
            availability: account.portfolio ? 'complete' : 'missing',
            fetchedAt: account.portfolio?.fetchedAt ?? null,
            freshForMs: portfolioFreshForMs,
            work:
              saved?.work.find(
                (work) => work.scopeKey === portfolioWorkKey(account.accountId),
              ) ?? null,
          }),
        })),
    };
  }

  async read(userId: string, accountId: string, observe = false) {
    const saved = await this.saved(userId, accountId, observe);
    const missing =
      saved.membership.availability === 'missing' ||
      saved.accounts.some((account) => !account.snapshot);
    const metadata = await this.metadata.savedMany(
      saved.accounts.flatMap((account) => account.snapshot?.positions ?? []),
    );
    const metadataDependencies = [...metadata.entries()].flatMap(
      ([key, source]) =>
        source.complete && source.data
          ? [
              investmentResourceMeta({
                resourceKey: `instrument/metadata/${key}`,
                dataRevision: source.dataRevision,
                availability: 'complete',
                fetchedAt: source.fetchedAt,
                freshForMs: tbankSourceTtlMs.metadata,
                work: null,
              }),
            ]
          : [],
    );
    const portfolios = saved.accounts.flatMap((account) =>
      account.snapshot
        ? [
            {
              ...account.snapshot,
              positions: account.snapshot.positions.map((position) => {
                const shared = metadata.get(metadataKey(position))?.data;
                return {
                  ...position,
                  ...(shared
                    ? {
                        displayName:
                          shared.name?.trim() ||
                          shared.ticker?.trim() ||
                          position.displayName,
                        ticker: shared.ticker?.trim() || position.ticker,
                        isin: shared.isin?.trim() || position.isin,
                        logoName:
                          shared.brand?.logoName?.trim() || position.logoName,
                      }
                    : {}),
                  ...(accountId === 'all'
                    ? { account: { id: account.id, name: account.name } }
                    : {}),
                };
              }),
            },
          ]
        : [],
    );
    const lastSuccessfulAt =
      // eslint-disable-next-line sonarjs/no-alphabetical-sort -- Code-unit ordering is intentional for canonical keys and ISO dates; locale collation would change persisted bindings.
      portfolios.map((portfolio) => portfolio.lastSuccessfulAt).sort()[0] ??
      saved.membership.lastSuccessfulAt;
    const data =
      missing || !lastSuccessfulAt
        ? null
        : {
            lastSuccessfulAt,
            isStale: portfolios.length
              ? portfolios.some((portfolio) => portfolio.isStale)
              : saved.membership.freshness !== 'fresh',
            totalsByCurrency: sumMoneyByCurrency(
              portfolios.flatMap((portfolio) => portfolio.totalsByCurrency),
            ),
            cashBalances: sumMoneyByCurrency(
              portfolios.flatMap((portfolio) => portfolio.cashBalances),
            ),
            positions: portfolios.flatMap((portfolio) => portfolio.positions),
            structure: portfolioStructure(
              portfolios.flatMap((portfolio) => portfolio.positions),
              portfolios.flatMap((portfolio) => portfolio.cashBalances),
            ),
          };
    const work =
      saved.accounts
        .map((account) => account.meta.sync)
        .find((work) => work.state === 'running') ??
        saved.accounts
          .map((account) => account.meta.sync)
          .find((work) => work.state === 'failed') ??
          saved.membership.sync;
    return {
      data,
      meta: investmentResourceMeta({
        resourceKey: `portfolio/${encodeURIComponent(accountId)}`,
        dataRevision: JSON.stringify([
          saved.accounts.map((account) => [
            account.id,
            account.meta.dataRevision,
          ]),
          metadataDependencies.map((dependency) => dependency.dataRevision),
        ]),
        availability: missing
          ? portfolios.length
            ? 'partial'
            : 'missing'
          : 'complete',
        fetchedAt: lastSuccessfulAt ? new Date(lastSuccessfulAt) : null,
        freshForMs: portfolios.length
          ? portfolioFreshForMs
          : membershipFreshForMs,
        work: work.id ? { ...work, id: work.id } : null,
        dependencies: [
          saved.membership,
          ...saved.accounts.map((account) => account.meta),
          ...metadataDependencies,
        ],
      }),
    };
  }

  request(userId: string, revision: string, accountId: string, force = false) {
    return this.queueSnapshot(userId, revision, accountId, force);
  }

  private queueSnapshot(
    userId: string,
    revision: string,
    accountId: string,
    force = false,
  ) {
    const reserved = this.reservations.then(async () => {
      const requested = await this.repository.reserve(
        userId,
        revision,
        accountId,
        this.processRunId,
        this.active,
        force,
      );
      if (!requested) return null;
      if (!requested.launch) {
        return {
          id: requested.work.id,
          done: this.pending.get(requested.work.attemptId) ?? Promise.resolve(),
        };
      }
      const { id, attemptId } = requested.work;
      this.active.add(attemptId);
      const done = new Promise<void>((resolve) => setImmediate(resolve))
        .then(async () => {
          const attempt = { id, attemptId, userId, revision };
          try {
            const token = await this.access.tokenForAttempt(attempt);
            if (!token) return;
            const [portfolio, positions] = await Promise.all([
              this.tbank.getPortfolio(token, accountId),
              this.tbank.getPositions(token, accountId),
            ]);
            const snapshot = await this.normalize(
              portfolio.positions ?? [],
              positions.money ?? [],
            );
            const published = await this.repository.publish(
              attempt,
              accountId,
              snapshot,
            );
            if (published) {
              for (const position of snapshot.positions) {
                await this.metadata.request(userId, revision, position);
              }
              if (!force) {
                await this.candles.prefetch(
                  userId,
                  revision,
                  snapshot.positions,
                );
              }
            }
          } catch (error) {
            if (
              error instanceof Error &&
              isTbankErrorCode(error.message, TbankErrorCode.InvalidToken)
            ) {
              await this.access.rejectMembershipToken(attempt);
            } else await this.repository.fail(attempt);
          }
        })
        .catch(() =>
          this.logger.event('error', 'tbank.portfolio.persistence_failed', {
            userId,
          }),
        )
        .finally(() => {
          this.active.delete(attemptId);
          this.pending.delete(attemptId);
        });
      this.pending.set(attemptId, done);
      return { id, done };
    });
    this.reservations = reserved.catch(() => undefined);
    return reserved;
  }

  async get(userId: string, accountId: string): Promise<PortfolioDto> {
    const result = await this.read(userId, accountId);
    if (!result.data) {
      throw new ServiceUnavailableException({ code: TbankErrorCode.Unavailable });
    }
    return result.data;
  }

  getAll(userId: string): Promise<PortfolioDto> {
    return this.get(userId, 'all');
  }

  private async normalize(
    upstream: {
      averagePositionPrice?: TbankMoney;
      currentPrice?: TbankMoney;
      figi?: string;
      instrumentType?: string;
      instrumentUid?: string;
      quantity?: { nano?: number | string; units?: number | string };
    }[],
    money: TbankMoney[],
  ): Promise<PortfolioDto> {
    const lastSuccessfulAt = new Date().toISOString().replace(/\.\d{3}Z$/, 'Z');
    const savedMetadata = await this.metadata.savedMany(upstream);
    const positions = upstream.map((position) =>
      this.position(
        position,
        savedMetadata.get(metadataKey(position))?.data ?? null,
      ),
    );
    const cash = new Map<string, ReturnType<typeof decimal>>();
    for (const value of money) {
      const currency = value.currency?.trim().toUpperCase();
      const amount = decimal(value);
      if (!currency || isZero(amount)) continue;
      cash.set(currency, add(cash.get(currency) ?? decimal({}), amount));
    }
    const cashBalances = [...cash].map(([currency, value]) => ({
      currency,
      value: formatDecimal(value),
    }));
    const normalizedPositions = positions.filter(
      (position): position is PortfolioPositionDto => position !== null,
    );
    return {
      lastSuccessfulAt,
      isStale: false,
      totalsByCurrency: totalsByCurrency(upstream, money),
      positions: normalizedPositions,
      cashBalances,
      structure: portfolioStructure(normalizedPositions, cashBalances),
    };
  }

  private position(
    position: Parameters<PortfolioService['normalize']>[0][number],
    metadata: null | TbankInstrument,
  ): null | PortfolioPositionDto {
    const quantity = decimal(position.quantity ?? {});
    if (isZero(quantity)) return null;
    const figi = position.figi?.trim() || null;
    const instrumentUid = position.instrumentUid?.trim() || null;
    const currentPrice = this.money(position.currentPrice);
    const averagePurchasePrice = this.money(position.averagePositionPrice);
    const currentValue = currentPrice && this.value(quantity, currentPrice);
    const absoluteReturn =
      currentPrice &&
      averagePurchasePrice &&
      currentPrice.currency === averagePurchasePrice.currency
        ? {
            currency: currentPrice.currency,
            value: formatDecimal(
              multiply(
                subtract(
                  decimalFromString(currentPrice.value),
                  decimalFromString(averagePurchasePrice.value),
                ),
                quantity,
              ),
            ),
          }
        : null;
    return {
      figi,
      instrumentUid,
      displayName:
        metadata?.name?.trim() ||
        metadata?.ticker?.trim() ||
        figi ||
        'Unknown instrument',
      type: types[position.instrumentType?.toLowerCase() ?? ''] ?? 'OTHER',
      quantity: formatDecimal(quantity),
      currentValue,
      averagePurchasePrice,
      absoluteReturn,
      isin: metadata?.isin?.trim() || null,
      ticker: metadata?.ticker?.trim() || null,
      logoName: metadata?.brand?.logoName?.trim() || null,
    };
  }

  private value(
    quantity: ReturnType<typeof decimal>,
    price: MoneyDto,
  ): MoneyDto {
    return {
      currency: price.currency,
      value: formatDecimal(multiply(quantity, decimalFromString(price.value))),
    };
  }

  private money(value: TbankMoney | undefined): MoneyDto | null {
    const currency = value?.currency?.trim().toUpperCase();
    return currency && value
      ? { currency, value: formatDecimal(decimal(value)) }
      : null;
  }
}