import { Injectable, NotFoundException, } from '@nestjs/common'; import { randomUUID } from 'node:crypto'; import { StructuredLogger } from '../../../../common/logging/structured-logger.service'; import { isTbankErrorCode, TbankErrorCode } from '../../../../integrations/tbank/tbank-error-code'; import { InstrumentMetadataService, type MetadataIdentity, metadataKey, TbankAccessService, TbankClient, TbankMembershipService, type TbankMoney, type TbankPortfolio, tbankSourceTtlMs, } from '../../../../integrations/tbank/tbank.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 { mapPortfolio } from '../infrastructure/tbank-portfolio.mapper'; import { PortfolioDto } from '../presentation/portfolio.dto'; const MISSING_DATA_REVISION = 0; const membershipFreshForMs = tbankSourceTtlMs.membership; @Injectable() export class PortfolioService { private readonly processRunId = randomUUID(); private readonly pending = new Map>(); private reservations: Promise = Promise.resolve(); constructor( private readonly access: TbankAccessService, private readonly tbank: TbankClient, private readonly metadata: InstrumentMetadataService, private readonly repository: PortfolioRepository, private readonly membership: TbankMembershipService, private readonly logger: StructuredLogger, ) {} async saved(userId: string, accountId: string, observe = false) { 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(); 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); } } } 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.pending, 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; const done = new Promise((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); } } } 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.pending.delete(attemptId); }); this.pending.set(attemptId, done); return { id, done }; }); this.reservations = reserved.catch(() => undefined); return reserved; } private async normalize( upstream: NonNullable, money: TbankMoney[], ): Promise { const lastSuccessfulAt = new Date().toISOString().replace(/\.\d{3}Z$/, 'Z'); const savedMetadata = await this.metadata.savedMany(upstream); return mapPortfolio(upstream, money, savedMetadata, lastSuccessfulAt); } }