import { Injectable, NotFoundException, ServiceUnavailableException, } from '@nestjs/common'; import { ConfigService } from '@nestjs/config'; import { ChannelCredentials, Client, ClientUnaryCall, loadPackageDefinition, Metadata, ServiceError, status, } from '@grpc/grpc-js'; import { loadSync } from '@grpc/proto-loader'; import { existsSync, readFileSync } from 'node:fs'; import { join } from 'node:path'; import { Environment } from '../config/env.validation'; import { StructuredLogger } from '../common/logging/structured-logger.service'; import type { TbankShare, TbankBond, TbankCoupon, TbankCandle, TbankLastPrice, TbankFundamentals, } from './tbank-market.types'; const requestDeadlineMs = 15_000; type Unary = ( request: Request, metadata: Metadata, options: { deadline: Date }, callback: (error: ServiceError | null, response: Response) => void, ) => ClientUnaryCall; type TbankClientService = Client & Record>; export type TbankDecimal = { units?: string | number; nano?: string | number }; export type TbankMoney = TbankDecimal & { currency?: string }; export type TbankAccount = { id: string; type: string; name: string; openedDate?: { seconds?: string | number }; }; export type TbankPortfolio = { positions?: { figi?: string; instrumentUid?: string; instrumentType?: string; ticker?: string; quantity?: TbankDecimal; currentPrice?: TbankMoney; averagePositionPrice?: TbankMoney; }[]; }; export type TbankPositions = { money?: TbankMoney[] }; export type TbankInstrument = { ticker?: string; name?: string; isin?: string; brand?: { logoName?: string | null }; }; export type TbankDividend = { dividendNet?: TbankMoney; recordDate?: TbankTimestamp; paymentDate?: TbankTimestamp; lastBuyDate?: TbankTimestamp; dividendType?: string; }; export type TbankTimestamp = { seconds?: string | number; nanos?: number }; export type TbankOperationItem = { cursor?: string; id?: string; name?: string; date?: TbankTimestamp; type?: string; description?: string; figi?: string; instrumentUid?: string; instrumentType?: string; ticker?: string; payment?: TbankMoney; price?: TbankMoney; commission?: TbankMoney; accruedInt?: TbankMoney; quantity?: string | number; quantityDone?: string | number; }; export type TbankOperationsByCursorRequest = { accountId: string; instrumentId?: string; from?: TbankTimestamp; to?: TbankTimestamp; cursor?: string; limit?: number; state: 'OPERATION_STATE_EXECUTED'; }; export type TbankOperationsByCursorResponse = { hasNext?: boolean; nextCursor?: string; items?: TbankOperationItem[]; }; type TbankNamespace = { tinkoff: { public: { invest: { api: { contract: { v1: Record } } } }; }; }; class RateLimiter { private readonly waiting: (() => void)[] = []; private starts = 0; private windowStartedAt = Date.now(); constructor(private readonly perSecond: number) {} get size(): number { return this.waiting.length; } async run(operation: () => Promise): Promise { await new Promise((resolve) => { this.waiting.push(resolve); this.drain(); }); try { return await operation(); } finally { this.drain(); } } private drain(): void { const now = Date.now(); if (now - this.windowStartedAt >= 1000) { this.windowStartedAt = now; this.starts = 0; } while (this.starts < this.perSecond && this.waiting.length) { this.starts += 1; this.waiting.shift()?.(); } if (this.waiting.length && this.starts === this.perSecond) { setTimeout( () => this.drain(), Math.max(1, 1000 - (now - this.windowStartedAt)), ); } } } @Injectable() export class TbankClient { private readonly queue: RateLimiter; private readonly users: TbankClientService; private readonly operations: TbankClientService; private readonly instruments: TbankClientService; private readonly sparseInstruments: TbankClientService; private readonly marketData: TbankClientService; constructor( config: ConfigService, private readonly logger: StructuredLogger, ) { const sourceProtoRoot = join(process.cwd(), 'src/tbank/proto/contracts'); const protoRoot = existsSync(sourceProtoRoot) ? sourceProtoRoot : join(process.cwd(), 'dist/tbank/proto/contracts'); const definition = loadSync( [ 'users.proto', 'operations.proto', 'instruments.proto', 'marketdata.proto', ].map((file) => join(protoRoot, file)), { includeDirs: [protoRoot], keepCase: false, longs: String, enums: String, defaults: true, oneofs: true, }, ); const services = ( loadPackageDefinition(definition) as unknown as TbankNamespace ).tinkoff.public.invest.api.contract.v1; const caPath = config.get('TBANK_CA_CERT_PATH'); const credentials = caPath ? ChannelCredentials.createSsl(readFileSync(caPath)) : ChannelCredentials.createSsl(); const endpoint = config.getOrThrow('TBANK_GRPC_ENDPOINT'); const create = (name: string) => new ( services[name] as new ( address: string, credentials: ChannelCredentials, ) => TbankClientService )(endpoint, credentials); this.users = create('UsersService'); this.operations = create('OperationsService'); this.instruments = create('InstrumentsService'); this.marketData = create('MarketDataService'); // Preserve field presence for fundamentals; the established clients retain their decoding contract. const sparse = ( loadPackageDefinition( loadSync(join(protoRoot, 'instruments.proto'), { includeDirs: [protoRoot], keepCase: false, longs: String, enums: String, defaults: false, oneofs: true, }), ) as unknown as TbankNamespace ).tinkoff.public.invest.api.contract.v1; this.sparseInstruments = new ( sparse.InstrumentsService as new ( address: string, credentials: ChannelCredentials, ) => TbankClientService )(endpoint, credentials); this.queue = new RateLimiter( config.getOrThrow('TBANK_RATE_LIMIT_PER_SECOND'), ); this.logger.info('tbank.client.configured'); } async validateToken(token: string): Promise { await this.getOpenAccounts(token); } getOpenAccounts(token: string): Promise { return this.call(token, this.users, 'getAccounts', { status: 'ACCOUNT_STATUS_OPEN', }).then( (response: { accounts?: TbankAccount[] }) => response.accounts ?? [], ); } getPortfolio(token: string, accountId: string): Promise { return this.call(token, this.operations, 'getPortfolio', { accountId }); } getPositions(token: string, accountId: string): Promise { return this.call(token, this.operations, 'getPositions', { accountId }); } getOperationsByCursor( token: string, request: TbankOperationsByCursorRequest, ): Promise { return this.call(token, this.operations, 'getOperationsByCursor', request); } getInstrumentByFigi( token: string, figi: string, ): Promise { return this.call(token, this.instruments, 'getInstrumentBy', { idType: 'INSTRUMENT_ID_TYPE_FIGI', id: figi, }).then( (response: { instrument?: TbankInstrument }) => response.instrument ?? null, ); } getInstrumentByUid( token: string, instrumentUid: string, ): Promise { return this.call(token, this.instruments, 'getInstrumentBy', { idType: 'INSTRUMENT_ID_TYPE_UID', id: instrumentUid, }).then( (response: { instrument?: TbankInstrument }) => response.instrument ?? null, ); } getDividends( token: string, instrumentId: string, from: string, to: string, ): Promise { return this.call(token, this.instruments, 'getDividends', { instrumentId, from: { seconds: String(utcSeconds(from)) }, to: { seconds: String(utcSeconds(to)) }, }).then( (response: { dividends?: TbankDividend[] }) => response.dividends ?? [], ); } findShares(token: string, query: string): Promise { return this.call<{ instruments?: TbankShare[] }>( token, this.instruments, 'findInstrument', { query, instrumentKind: 'INSTRUMENT_TYPE_SHARE', }, ).then((response) => response.instruments ?? []); } findBonds(token: string, query: string): Promise { return this.call<{ instruments?: TbankBond[] }>( token, this.sparseInstruments, 'findInstrument', { query, instrumentKind: 'INSTRUMENT_TYPE_BOND', }, ).then((response) => response.instruments ?? []); } getBond(token: string, uid: string): Promise { return this.call<{ instrument?: TbankBond }>( token, this.sparseInstruments, 'bondBy', { idType: 'INSTRUMENT_ID_TYPE_UID', id: uid, }, ).then((response) => response.instrument ?? null); } getBondCoupons( token: string, uid: string, from: string, to: string, ): Promise { return this.call<{ events?: TbankCoupon[] }>( token, this.sparseInstruments, 'getBondCoupons', { instrumentId: uid, from: { seconds: String(utcSeconds(from)) }, to: { seconds: String(utcSeconds(to)) }, }, ).then((response) => response.events ?? []); } getShare(token: string, uid: string): Promise { return this.call<{ instrument?: TbankShare }>( token, this.instruments, 'shareBy', { idType: 'INSTRUMENT_ID_TYPE_UID', id: uid, }, ).then((response) => response.instrument ?? null); } getAssetDescription(token: string, assetUid: string): Promise { return this.call<{ asset?: { description?: string } }>( token, this.instruments, 'getAssetBy', { id: assetUid, }, ).then((response) => response.asset?.description?.trim() || null); } getFundamentals( token: string, assetUid: string, ): Promise { return this.call<{ fundamentals?: TbankFundamentals[] }>( token, this.sparseInstruments, 'getAssetFundamentals', { assets: [assetUid], }, ).then( (response) => response.fundamentals?.find((item) => item.assetUid === assetUid) ?? null, ); } getLastPrices(token: string, uids: string[]): Promise { return this.call<{ lastPrices?: TbankLastPrice[] }>( token, this.marketData, 'getLastPrices', { instrumentId: uids }, ).then((response) => response.lastPrices ?? []); } getLastPrice(token: string, uid: string): Promise { return this.getLastPrices(token, [uid]).then( (prices) => prices.find((item) => item.instrumentUid === uid) ?? null, ); } getTradingStatus(token: string, uid: string): Promise { return this.call<{ tradingStatus?: string }>( token, this.marketData, 'getTradingStatus', { instrumentId: uid, }, ).then((response) => response.tradingStatus || null); } getCandles( token: string, uid: string, from: Date, to: Date, ): Promise { return this.call<{ candles?: TbankCandle[] }>( token, this.marketData, 'getCandles', { instrumentId: uid, from: { seconds: String(Math.floor(from.valueOf() / 1000)) }, to: { seconds: String(Math.floor(to.valueOf() / 1000)) }, interval: 'CANDLE_INTERVAL_DAY', limit: 2400, }, ).then((response) => response.candles ?? []); } private async call( token: string, client: TbankClientService, method: string, request: object, ): Promise { if (this.queue.size >= 50) { this.logger.event('error', 'tbank.rpc.failed', { operation: method, code: 'TBANK_UNAVAILABLE', attempts: 0, }); throw new ServiceUnavailableException({ code: 'TBANK_UNAVAILABLE' }); } for (let attempts = 1; attempts <= 2; attempts += 1) { const startedAt = performance.now(); try { const response = await this.queue.run(() => this.request(token, client, method, request), ); this.logger.event('debug', 'tbank.rpc.completed', { operation: method, attempts, durationMs: Math.round(performance.now() - startedAt), }); return response; } catch (error) { const grpcCode = (error as ServiceError).code; const retryable = [ status.UNAVAILABLE, status.DEADLINE_EXCEEDED, status.RESOURCE_EXHAUSTED, ].includes(grpcCode); if (retryable && attempts < 2) { this.logger.event('warn', 'tbank.rpc.retry', { operation: method, grpcCode, attempts, durationMs: Math.round(performance.now() - startedAt), }); await new Promise((resolve) => setTimeout(resolve, 1000)); continue; } this.logger.event('error', 'tbank.rpc.failed', { operation: method, grpcCode, attempts, durationMs: Math.round(performance.now() - startedAt), }); if ( [status.UNAUTHENTICATED, status.PERMISSION_DENIED].includes(grpcCode) ) throw new Error('INVALID_TBANK_TOKEN'); if (grpcCode === status.NOT_FOUND && method === 'shareBy') throw new NotFoundException(); throw new ServiceUnavailableException({ code: 'TBANK_UNAVAILABLE' }); } } throw new ServiceUnavailableException({ code: 'TBANK_UNAVAILABLE' }); } private request( token: string, client: TbankClientService, method: string, request: object, ): Promise { return new Promise((resolve, reject) => { const metadata = new Metadata(); metadata.set('Authorization', `Bearer ${token}`); client[method]( request as never, metadata, { deadline: new Date(Date.now() + requestDeadlineMs) }, (error, response) => (error ? reject(error) : resolve(response)), ); }); } } function utcSeconds(date: string): number { return Math.floor(Date.parse(`${date}T00:00:00Z`) / 1000); }