import { ConflictException, Injectable, NotFoundException, ServiceUnavailableException, } from '@nestjs/common'; import { randomUUID } from 'node:crypto'; import type { InvestmentResourceMetaDto } from '../investment-resource.dto'; import { ISO_DATE_LENGTH, ISO_DATE_START } from '../../../common/time'; import { TbankErrorCode } from '../../../integrations/tbank/tbank-error-code'; import { InstrumentMetadataService, TbankAccessService, TbankMembershipService, type TbankSyncAttempt, } from '../../../integrations/tbank/tbank.api'; import { BrokerageAccountsService } from '../brokerage-accounts/brokerage-accounts-read.api'; import { IncomeAnalyticsService } from '../income/income-read.api'; import { BondIncomeService, StockIncomeService, } from '../instruments/instrument-income.api'; import { CandleService, CatalogDiscoveryService, CatalogEnrichmentService, } from '../instruments/instruments.api'; import { investmentResourceMeta } from '../investment-resource.mapper'; import { planInvestmentScreen } from '../investment-sync-plan'; import { BrokerageOperationsService, BrokerageOperationsSyncService, type OperationInterval, } from '../operations/operations.api'; import { PortfolioService } from '../portfolio/portfolio.api'; import { UpcomingEventsService } from '../upcoming-events/upcoming-events-read.api'; import { normalizeScreen } from './tbank-screen'; import { InvestmentScreenDto, TbankSyncStatusDto } from './tbank-sync.dto'; import { type ScreenIntent, TbankSyncRepository, } from './tbank-sync.repository'; const FOREGROUND_CANDLE_PRIORITY = 0; const EXCLUSIVE_END_OFFSET_MS = 1; const PENDING_CHILD_COUNT = 0; const COMPLETED_CHILD_COUNT = 1; @Injectable() export class TbankSyncService { private readonly runId = randomUUID(); private readonly startedAt = new Date(); private reservations: Promise = Promise.resolve(); constructor( private readonly repository: TbankSyncRepository, private readonly access: TbankAccessService, private readonly membership: TbankMembershipService, private readonly portfolio: PortfolioService, private readonly operations: BrokerageOperationsSyncService, private readonly ledger: BrokerageOperationsService, private readonly metadata: InstrumentMetadataService, private readonly discovery: CatalogDiscoveryService, private readonly sources: CatalogEnrichmentService, private readonly candles: CandleService, private readonly accounts: BrokerageAccountsService, private readonly income: IncomeAnalyticsService, private readonly events: UpcomingEventsService, private readonly stockIncome: StockIncomeService, private readonly bondIncome: BondIncomeService, ) {} async request(userId: string, input: InvestmentScreenDto) { const scope = normalizeScreen(input); const connection = await this.access.local(userId, scope.accountId); const reservation = this.reservations.then(() => this.repository.reserve(userId, connection.revision, scope, this.runId), ); this.reservations = reservation.catch(() => undefined); const reserved = await reservation; if (!reserved) throw new ServiceUnavailableException(); const intent = reserved.work.intent as unknown as ScreenIntent; if (reserved.launch) { const attempt = { userId, revision: connection.revision, id: reserved.work.id, attemptId: reserved.work.attemptId, }; setImmediate(() => { void this.dispatch(attempt, intent); }); } return { id: reserved.work.id, ...intent }; } private async dispatch(attempt: TbankSyncAttempt, intent: ScreenIntent) { const children: string[] = []; const add = async (ids: (null | string | undefined)[]) => { children.push(...ids.filter((id): id is string => Boolean(id))); await this.repository.progress(attempt, children, false); }; try { const { userId, revision } = attempt; const { scope } = intent; const now = new Date(intent.now); const member = await this.membership.request(userId, revision, true); await add([member?.id]); await member?.done; const current = await this.access.local(userId, scope.accountId); if (current.revision !== revision) return; const accountScope = scope.accountId ?? 'all'; const selectedAccounts = current.accounts.filter( (account) => accountScope === 'all' || account.accountId === accountScope, ); if ( ['account', 'accounts', 'bond', 'events', 'stock'].includes( scope.screen, ) ) { for (const account of scope.screen === 'stock' || scope.screen === 'bond' ? current.accounts : selectedAccounts) { const work = await this.portfolio.request( userId, revision, account.accountId, true, ); await add([work?.id]); await work?.done; } } const saved = await this.portfolio.read( userId, scope.screen === 'stock' || scope.screen === 'bond' ? 'all' : accountScope, true, ); const positions = saved.data?.positions ?? []; const instruments = [ ...new Map( positions.flatMap((position) => position.instrumentUid && (position.type === 'STOCK' || position.type === 'BOND') ? [ [ position.instrumentUid, { uid: position.instrumentUid, type: position.type, historyFrom: null, }, ] as const, ] : [], ), ).values(), ]; const plan = planInvestmentScreen( { ...scope, candleRange: scope.candleRange ? { from: new Date(scope.candleRange.from), to: new Date(scope.candleRange.to), } : undefined, }, { accounts: selectedAccounts, instruments: scope.screen === 'stocks' || scope.screen === 'bonds' ? [] : instruments, membershipKnown: Boolean(current.membershipAt), }, now, true, ); const intervals: OperationInterval[] = plan.sources.flatMap((source) => source.kind === 'operations' && source.accountId && source.from && source.to ? [ { accountId: source.accountId, from: source.from, to: source.to, ...(source.instrumentUid ? { instrumentUid: source.instrumentUid } : {}), }, ] : [], ); if (intervals.length) { await add([ await this.operations.request( userId, intervals, true, 'screen', revision, ), ]); } for (const source of plan.sources) { const uid = source.instrumentUid; const type = source.instrumentType === 'BOND' ? 'bond' : 'share'; if (source.kind === 'catalog') { await add( await this.discovery.refresh(userId, revision, { instrumentType: type, query: source.query ?? '', }), ); } if (!uid) continue; if (source.kind === 'metadata') { await add([ await this.metadata.request(userId, revision, { instrumentUid: uid, }), ]); } if (source.kind === 'quote') { await add( await this.sources.requestMany( userId, revision, type, [uid], true, [ type === 'bond' ? 'bond-current' : 'detail', 'last-price', 'trading-status', 'comparison', ], ), ); } if (source.kind === 'company') { await add( await this.sources.requestMany( userId, revision, type, [uid], true, ['description', 'fundamentals'], ), ); } if (source.kind === 'metrics') { await add( await this.sources.requestMany( userId, revision, type, [uid], true, ['coupons', 'yield', 'rating'], ), ); } if (source.kind === 'calendar' && source.from && source.to) { await add( await this.sources.requestMany( userId, revision, type, [uid], true, [ type === 'share' ? 'dividends' : scope.screen === 'bond' ? 'coupons' : 'bond-events', ], { from: source.from.toISOString().slice(ISO_DATE_START, ISO_DATE_LENGTH), to: source.to.toISOString().slice(ISO_DATE_START, ISO_DATE_LENGTH), ticker: positions.find((position) => position.instrumentUid === uid) ?.ticker ?? undefined, }, ), ); } } if ( (scope.screen === 'stock' || scope.screen === 'bond') && scope.instrumentUid ) { await add([ await this.candles.request( userId, revision, { uid: scope.instrumentUid, instrumentType: scope.screen === 'bond' ? 'bond' : 'share', period: scope.candlePeriod ?? 'month', ...(scope.candleRange ? { range: { from: new Date(scope.candleRange.from), to: new Date(scope.candleRange.to), }, } : {}), }, now, true, FOREGROUND_CANDLE_PRIORITY, ), ]); } await this.repository.progress(attempt, children, true); } catch { await this.repository .progress(attempt, children, true, TbankErrorCode.Unavailable) .catch(() => undefined); } } async status(userId: string, id: string): Promise { const connection = await this.access .local(userId) .catch((error: unknown) => { if (error instanceof ConflictException) throw new NotFoundException(); throw error; }); const saved = await this.repository.read(userId, connection.revision, id); if (!saved) throw new NotFoundException(); if (!saved.work.scopeKey.startsWith('screen:')) { return this.sourceStatus(userId, saved.work); } const intent = saved.work.intent as unknown as ScreenIntent; await this.access.local(userId, intent.scope.accountId); const children = await Promise.all( saved.children.map((child) => child.state === 'superseded' ? this.operations.observe(userId, child.id) : Promise.resolve(child), ), ); const running = children.some((child) => child.state === 'running'); const failed = saved.work.errorCode ?? children.find((child) => child.errorCode)?.errorCode ?? null; const state = saved.work.processRunId !== this.runId && (!saved.progress.dispatched || running) ? 'cancelled' : !saved.progress.dispatched || running ? 'running' : failed ? 'failed' : children.some((child) => child.state === 'cancelled') ? 'cancelled' : 'succeeded'; const resources = await this.observe(userId, intent.scope); if ( (await this.access.local(userId, intent.scope.accountId)).revision !== connection.revision ) { throw new NotFoundException(); } return { id, ...intent, state, errorCode: failed, progress: { completed: children.filter((child) => child.state !== 'running').length, total: children.length, discovering: !saved.progress.dispatched, }, resources, }; } private async sourceStatus( userId: string, work: NonNullable>>['work'], ): Promise { const resources: InvestmentResourceMetaDto[] = []; const input = work.intent as Record; const add = async (read: Promise<{ meta: InvestmentResourceMetaDto }>) => { resources.push((await read).meta); }; if (work.scopeKey === 'membership') { await add(this.accounts.list(userId, true)); } else if ( input.source === 'portfolio' && typeof input.accountId === 'string' ) { await add(this.portfolio.read(userId, input.accountId, true)); } else if (input.source === 'catalog') { const type = input.instrumentType === 'bond' ? 'bond' : 'share'; if (typeof input.query === 'string') { await add( this.discovery.search( userId, { instrumentType: type, query: input.query, exactTicker: input.exactTicker === true, }, true, ), ); } else if ( typeof input.uid === 'string' && typeof input.kind === 'string' ) { const range = typeof input.rangeKey === 'string' && input.rangeKey.startsWith('[') ? (JSON.parse(input.rangeKey) as string[]) : null; const sources = await this.sources.readMany( userId, type, [input.uid], true, [ input.kind as Parameters< CatalogEnrichmentService['readMany'] >[4] extends (infer K)[] | undefined ? K : never, ], range ? { from: range[0], to: range[1] } : undefined, ); resources.push(...sources.map((source) => source.meta)); } } else if (input.source === 'metadata') { const identity = { instrumentUid: typeof input.instrumentUid === 'string' ? input.instrumentUid : null, figi: typeof input.figi === 'string' ? input.figi : null, }; const source = await this.metadata.read(userId, identity, true); resources.push( investmentResourceMeta({ resourceKey: work.scopeKey, dataRevision: source?.dataRevision ?? '0', availability: source?.complete ? 'complete' : 'missing', fetchedAt: source?.fetchedAt ?? null, freshForMs: Number.POSITIVE_INFINITY, work, }), ); } else if ( work.scopeKey.startsWith('candles:') && typeof input.uid === 'string' ) { await add( this.candles.read( userId, { uid: input.uid, instrumentType: input.instrumentType === 'bond' ? 'bond' : 'share', period: input.period as 'all' | 'halfYear' | 'month' | 'year', }, true, ), ); } else if ( work.scopeKey.startsWith('operations:') && Array.isArray(input.intervals) ) { for (const key of input.intervals as string[]) { const [accountId, from, end, instrumentUid] = JSON.parse( key, ) as string[]; await add( this.ledger.read(userId, accountId, { from, to: new Date(Date.parse(end) - EXCLUSIVE_END_OFFSET_MS).toISOString(), ...(instrumentUid ? { instrumentUid } : {}), sync: 'observe', }), ); } } const observed = work.scopeKey.startsWith('operations:') ? await this.operations.observe(userId, work.id) : work; const state = observed.state === 'running' ? observed.updatedAt < this.startedAt ? 'cancelled' : 'running' : observed.state === 'failed' ? 'failed' : observed.state === 'cancelled' ? 'cancelled' : 'succeeded'; if ((await this.access.local(userId)).revision !== work.revision) { throw new NotFoundException(); } return { id: work.id, scope: null, now: work.createdAt.toISOString(), state, errorCode: observed.errorCode, progress: { completed: state === 'running' ? PENDING_CHILD_COUNT : COMPLETED_CHILD_COUNT, total: 1, discovering: false, }, resources, }; } private async observe(userId: string, scope: InvestmentScreenDto) { const resources: InvestmentResourceMetaDto[] = []; const add = async (read: Promise<{ meta: InvestmentResourceMetaDto }>) => { resources.push((await read).meta); }; const account = scope.accountId ?? 'all'; if (scope.screen === 'accounts') { await add(this.accounts.list(userId, true)); } if (scope.screen === 'account') { await add(this.portfolio.read(userId, account, true)); await add(this.income.summary(userId, account, true)); await add(this.events.get(userId, account, { sync: 'observe' })); } if (['account', 'bond', 'operations', 'stock'].includes(scope.screen)) { await add( this.ledger.read(userId, account, { from: scope.from, to: scope.to, instrumentUid: scope.instrumentUid, categories: scope.categories, limit: scope.limit, sync: 'observe', }), ); } if (scope.screen === 'income') { await add( this.income.analytics(userId, account, scope.from, scope.to, true), ); } if (scope.screen === 'events') { await add( this.events.get(userId, account, { from: scope.from, to: scope.to, sync: 'observe', }), ); } const type = scope.screen === 'bond' || scope.screen === 'bonds' ? 'bond' : 'share'; if (scope.screen === 'stocks' || scope.screen === 'bonds') { await add( this.discovery.search( userId, { instrumentType: type, query: scope.query ?? '' }, true, ), ); } if ( (scope.screen === 'stock' || scope.screen === 'bond') && scope.instrumentUid ) { const uid = scope.instrumentUid; await add(this.discovery.detail(userId, type, uid, true)); await add(this.sources.section(userId, type, uid, 'quote', true)); await add( this.sources.section( userId, type, uid, type === 'share' ? 'company' : 'metrics', true, ), ); await add( this.candles.read( userId, { uid, instrumentType: type, period: scope.candlePeriod ?? 'month', ...(scope.candleRange ? { range: { from: new Date(scope.candleRange.from), to: new Date(scope.candleRange.to), }, } : {}), }, true, ), ); await add( type === 'share' ? this.stockIncome.readPosition(userId, uid, true) : this.bondIncome.readPosition(userId, uid, true), ); await add( type === 'share' ? this.stockIncome.dividends(userId, uid, true) : this.bondIncome.coupons(userId, uid, true), ); } return resources; } }