From eaf2f67b34b2f8b45a9f5d53c544b91304b333d3 Mon Sep 17 00:00:00 2001 From: Sergey Krylov Date: Wed, 9 Sep 2026 20:36:31 +0300 Subject: [PATCH] feat: aggregate portfolio data across accounts Add authenticated all-account portfolio, operations, income and event endpoints. Preserve source accounts, exact currency totals and scoped cursor pagination. --- src/tbank/account-reference.dto.ts | 10 + src/tbank/all-accounts.spec.ts | 342 +++++++++++++++++++ src/tbank/brokerage-operation.dto.ts | 4 + src/tbank/brokerage-operations.controller.ts | 9 + src/tbank/brokerage-operations.service.ts | 230 ++++++++++++- src/tbank/income-analytics.controller.ts | 15 + src/tbank/income-analytics.service.ts | 2 +- src/tbank/portfolio-totals.ts | 27 +- src/tbank/portfolio.controller.ts | 6 + src/tbank/portfolio.dto.ts | 6 +- src/tbank/portfolio.service.ts | 120 ++++--- src/tbank/upcoming-events.controller.ts | 6 + src/tbank/upcoming-events.dto.ts | 6 +- src/tbank/upcoming-events.service.ts | 108 ++++-- test/all-accounts.e2e-spec.ts | 101 ++++++ 15 files changed, 909 insertions(+), 83 deletions(-) create mode 100644 src/tbank/account-reference.dto.ts create mode 100644 src/tbank/all-accounts.spec.ts create mode 100644 test/all-accounts.e2e-spec.ts diff --git a/src/tbank/account-reference.dto.ts b/src/tbank/account-reference.dto.ts new file mode 100644 index 0000000..591493e --- /dev/null +++ b/src/tbank/account-reference.dto.ts @@ -0,0 +1,10 @@ +import { ApiProperty } from '@nestjs/swagger'; + +/** Source account of a row in an all-accounts response. */ +export class AccountReferenceDto { + @ApiProperty() + id: string; + + @ApiProperty() + name: string; +} diff --git a/src/tbank/all-accounts.spec.ts b/src/tbank/all-accounts.spec.ts new file mode 100644 index 0000000..26ef97c --- /dev/null +++ b/src/tbank/all-accounts.spec.ts @@ -0,0 +1,342 @@ +import { + BadRequestException, + ConflictException, + ServiceUnavailableException, +} from '@nestjs/common'; + +jest.mock('./tbank-access.service', () => ({ TbankAccessService: class {} })); + +import { BrokerageOperationsService } from './brokerage-operations.service'; +import { IncomeAnalyticsService } from './income-analytics.service'; +import { PortfolioService } from './portfolio.service'; +import { UpcomingEventsService } from './upcoming-events.service'; +import type { TbankConnectionContext } from './tbank-access.service'; + +function setup() { + const context = { + token: 'token', + generation: 'first', + accounts: [ + { id: 'a', name: 'Основной', openedOn: '2020-01-01' }, + { id: 'b', name: 'ИИС', openedOn: '2026-01-01' }, + ], + }; + const access = { + withSupportedAccounts: jest.fn( + (_user: string, handler: (value: typeof context) => unknown) => + handler(context), + ), + withAccountForConnection: jest.fn( + ( + _connection: TbankConnectionContext, + _id: string, + handler: () => unknown, + ) => handler(), + ), + }; + const tbank = { + getOperationsByCursor: jest.fn(), + getPortfolio: jest.fn().mockResolvedValue({ positions: [] }), + getPositions: jest.fn().mockResolvedValue({ money: [] }), + getDividends: jest.fn().mockResolvedValue([]), + getInstrumentByFigi: jest.fn().mockResolvedValue({ name: 'Облигация' }), + }; + const metadata = { resolve: jest.fn().mockResolvedValue(null) }; + const requestContext = { set: jest.fn() }; + const operations = new BrokerageOperationsService( + access as never, + tbank as never, + requestContext as never, + metadata as never, + ); + const portfolio = new PortfolioService( + access as never, + tbank as never, + requestContext as never, + metadata as never, + ); + const moex = { bondEvents: jest.fn().mockResolvedValue([]) }; + const events = new UpcomingEventsService( + access as never, + tbank as never, + moex as never, + requestContext as never, + { info: jest.fn(), event: jest.fn() } as never, + ); + return { + context, + access, + tbank, + operations, + portfolio, + events, + moex, + income: new IncomeAnalyticsService(operations), + }; +} + +const operation = ( + id: string, + second: number, + type = 'OPERATION_TYPE_COUPON', + value = 1, +) => ({ + id, + date: { seconds: String(second) }, + type, + payment: { currency: 'RUB', units: value }, +}); +const query = { from: '2025-01-01', to: '2026-12-31', limit: 2 }; + +describe('All accounts', () => { + afterEach(() => jest.useRealTimers()); + + it('merges exact amounts by currency and keeps equal instruments on different accounts', async () => { + const { portfolio, tbank } = setup(); + tbank.getPortfolio.mockResolvedValue({ + positions: [ + { + figi: 'SAME', + instrumentType: 'share', + quantity: { units: 1 }, + currentPrice: { currency: 'RUB', nano: 100000000 }, + }, + ], + }); + tbank.getPositions.mockImplementation((_token: string, id: string) => ({ + money: [{ currency: id === 'a' ? 'RUB' : 'USD', nano: 200000000 }], + })); + const result = await portfolio.getAll('user'); + expect(result.totalsByCurrency).toEqual([ + { currency: 'RUB', value: '0.4' }, + { currency: 'USD', value: '0.2' }, + ]); + expect(result.cashBalances).toEqual([ + { currency: 'RUB', value: '0.2' }, + { currency: 'USD', value: '0.2' }, + ]); + expect(result.positions.map((position) => position.account?.id)).toEqual([ + 'a', + 'b', + ]); + }); + + it('rejects missing access and a partial portfolio instead of publishing incomplete totals', async () => { + const { portfolio, access, tbank } = setup(); + access.withSupportedAccounts.mockImplementationOnce(() => { + throw new ConflictException({ code: 'TBANK_NOT_CONNECTED' }); + }); + await expect(portfolio.getAll('user')).rejects.toBeInstanceOf( + ConflictException, + ); + expect(tbank.getPortfolio).not.toHaveBeenCalled(); + tbank.getPortfolio.mockRejectedValueOnce(new ServiceUnavailableException()); + await expect(portfolio.getAll('user')).rejects.toBeInstanceOf( + ServiceUnavailableException, + ); + }); + + it('marks an aggregate stale and uses its oldest snapshot timestamp', async () => { + jest.useFakeTimers().setSystemTime(new Date('2026-09-01T10:00:00Z')); + const { portfolio, tbank } = setup(); + const first = await portfolio.getAll('user'); + jest.setSystemTime(new Date('2026-09-01T10:01:00Z')); + tbank.getPortfolio.mockRejectedValueOnce(new ServiceUnavailableException()); + const result = await portfolio.getAll('user'); + expect(result.isStale).toBe(true); + expect(result.lastSuccessfulAt).toBe(first.lastSuccessfulAt); + }); + + it('paginates interleaved streams without dropping buffered rows, preserving duplicate IDs', async () => { + const { operations, tbank } = setup(); + tbank.getOperationsByCursor.mockImplementation( + ( + _token: string, + { accountId, cursor }: { accountId: string; cursor?: string }, + ) => { + if (accountId === 'a') + return cursor + ? { items: [operation('last', 3)] } + : { + items: [operation('same', 9), operation('a2', 5)], + hasNext: true, + nextCursor: 'a-next', + }; + return { items: [operation('same', 8), operation('b2', 4)] }; + }, + ); + const first = await operations.getAll('user', query); + const second = await operations.getAll('user', { + ...query, + cursor: first.nextCursor!, + }); + const repeated = await operations.getAll('user', { + ...query, + cursor: first.nextCursor!, + }); + const third = await operations.getAll('user', { + ...query, + cursor: second.nextCursor!, + }); + expect(first.items.map((item) => [item.id, item.account?.id])).toEqual([ + ['same', 'a'], + ['same', 'b'], + ]); + expect(second.items.map((item) => item.id)).toEqual(['a2', 'b2']); + expect(repeated).toEqual(second); + expect(third.items.map((item) => item.id)).toEqual(['last']); + expect(third.hasNext).toBe(false); + }); + + it('continues through empty filtered pages and breaks ties by account', async () => { + const { operations, tbank } = setup(); + tbank.getOperationsByCursor.mockImplementation( + (_token: string, { cursor }: { cursor?: string }) => + cursor + ? { items: [operation('income', 5)] } + : { + items: [operation('fee', 8, 'OPERATION_TYPE_BROKER_FEE')], + hasNext: true, + nextCursor: 'next', + }, + ); + const page = await operations.getAll('user', { + ...query, + categories: ['income'], + }); + expect(page.items.map((item) => item.account?.id)).toEqual(['a', 'b']); + expect(page.items.every((item) => item.category === 'income')).toBe(true); + }); + + it('rejects damaged cursor chains and partial streams', async () => { + const { operations, tbank } = setup(); + tbank.getOperationsByCursor.mockResolvedValue({ + items: [], + hasNext: true, + nextCursor: 'loop', + }); + await expect(operations.getAll('user', query)).rejects.toBeInstanceOf( + ServiceUnavailableException, + ); + tbank.getOperationsByCursor.mockRejectedValue( + new ServiceUnavailableException(), + ); + await expect( + operations.getAll('another-user', query), + ).rejects.toBeInstanceOf(ServiceUnavailableException); + }); + + it('binds continuations to user, connection, accounts and filters', async () => { + const { operations, context, tbank } = setup(); + tbank.getOperationsByCursor.mockResolvedValue({ + items: [operation('first', 8), operation('last', 4)], + }); + const first = await operations.getAll('user', query); + const next = { ...query, cursor: first.nextCursor! }; + await expect(operations.getAll('other', next)).rejects.toBeInstanceOf( + BadRequestException, + ); + await expect( + operations.getAll('user', { ...next, categories: ['fee'] }), + ).rejects.toBeInstanceOf(BadRequestException); + context.generation = 'replaced'; + await expect(operations.getAll('user', next)).rejects.toBeInstanceOf( + BadRequestException, + ); + context.generation = 'first'; + context.accounts.pop(); + await expect(operations.getAll('user', next)).rejects.toBeInstanceOf( + BadRequestException, + ); + expect(tbank.getOperationsByCursor).toHaveBeenCalledTimes(2); + }); + + it('collects income across the same cohort and computes comparisons from all receipts', async () => { + const { income, tbank } = setup(); + tbank.getOperationsByCursor.mockImplementation( + ( + _token: string, + { accountId, from }: { accountId: string; from: { seconds: string } }, + ) => { + const year = new Date(Number(from.seconds) * 1000).getUTCFullYear(); + const second = + Date.parse( + year === 2025 ? '2025-08-15T12:00:00Z' : '2026-08-15T12:00:00Z', + ) / 1000; + return { + items: + year === 2025 && accountId === 'b' + ? [] + : [ + operation( + 'same', + second, + 'OPERATION_TYPE_COUPON', + year === 2025 ? 2 : 3, + ), + ], + }; + }, + ); + const result = await income.analytics( + 'user', + 'all', + '2026-08-01', + '2026-08-31', + ); + expect(result.totalsByCurrency[0].total.value).toBe('6'); + expect( + result.yearComparison?.totalsByCurrency[0].compared.total.value, + ).toBe('2'); + expect( + result.yearComparison?.totalsByCurrency[0].difference.total.value, + ).toBe('4'); + expect(result.entries.map((entry) => entry.account?.id)).toEqual([ + 'a', + 'b', + ]); + expect( + new Set(result.entries.map((entry) => entry.instrumentKey)).size, + ).toBe(2); + }); + + it('preserves separate future events and fails the section when one source fails', async () => { + const { events, tbank, moex } = setup(); + const today = new Date().toISOString().slice(0, 10); + tbank.getPortfolio.mockResolvedValue({ + positions: [ + { + figi: 'BOND', + ticker: 'BOND', + instrumentType: 'bond', + quantity: { units: 2 }, + }, + ], + }); + moex.bondEvents.mockResolvedValue([ + { type: 'coupon', date: today, value: '0.1', currency: 'RUB' }, + ]); + const result = await events.getAll('user'); + expect(result.events.map((event) => event.account?.id)).toEqual(['a', 'b']); + expect(result.incomeTotalsByCurrency).toEqual([ + { currency: 'RUB', value: '0.4' }, + ]); + tbank.getPortfolio + .mockResolvedValueOnce({ positions: [] }) + .mockRejectedValueOnce(new ServiceUnavailableException()); + await expect(events.getAll('user')).rejects.toBeInstanceOf( + ServiceUnavailableException, + ); + }); + + it('returns valid empty results for an empty account set', async () => { + const { context, portfolio, operations, income, events, tbank } = setup(); + context.accounts = []; + expect((await portfolio.getAll('user')).positions).toEqual([]); + expect((await operations.getAll('user', query)).hasNext).toBe(false); + expect((await income.summary('user', 'all')).totalsByCurrency).toEqual([]); + expect((await events.getAll('user')).events).toEqual([]); + expect(tbank.getOperationsByCursor).not.toHaveBeenCalled(); + expect(tbank.getPortfolio).not.toHaveBeenCalled(); + }); +}); diff --git a/src/tbank/brokerage-operation.dto.ts b/src/tbank/brokerage-operation.dto.ts index 418136b..1eb5ce1 100644 --- a/src/tbank/brokerage-operation.dto.ts +++ b/src/tbank/brokerage-operation.dto.ts @@ -12,6 +12,7 @@ import { Min, } from 'class-validator'; import { MoneyDto } from './portfolio.dto'; +import { AccountReferenceDto } from './account-reference.dto'; export const operationCategories = [ 'trade', @@ -59,6 +60,9 @@ export class BrokerageOperationsQueryDto { } export class BrokerageOperationDto { + @ApiPropertyOptional({ type: AccountReferenceDto }) + account?: AccountReferenceDto; + @ApiProperty() id: string; diff --git a/src/tbank/brokerage-operations.controller.ts b/src/tbank/brokerage-operations.controller.ts index 85f97f2..09b9ac3 100644 --- a/src/tbank/brokerage-operations.controller.ts +++ b/src/tbank/brokerage-operations.controller.ts @@ -18,6 +18,15 @@ import { BrokerageOperationsService } from './brokerage-operations.service'; export class BrokerageOperationsController { constructor(private readonly operations: BrokerageOperationsService) {} + @Get('all/operations') + @ApiOkResponse({ type: BrokerageOperationsPageDto }) + all( + @Req() request: Request & { user: AccessPayload }, + @Query() query: BrokerageOperationsQueryDto, + ) { + return this.operations.getAll(request.user.sub, query); + } + @Get(':accountId/operations') @ApiOkResponse({ type: BrokerageOperationsPageDto }) get( diff --git a/src/tbank/brokerage-operations.service.ts b/src/tbank/brokerage-operations.service.ts index a4fdadc..d2405d4 100644 --- a/src/tbank/brokerage-operations.service.ts +++ b/src/tbank/brokerage-operations.service.ts @@ -1,4 +1,11 @@ -import { Injectable, ServiceUnavailableException } from '@nestjs/common'; +import { + BadRequestException, + Injectable, + ServiceUnavailableException, +} from '@nestjs/common'; +import { randomUUID } from 'node:crypto'; +import { AccountReferenceDto } from './account-reference.dto'; +import { BrokerageOperationDto } from './brokerage-operation.dto'; import { LRUCache } from 'lru-cache'; import { RequestContextService } from '../common/logging/request-context.service'; import { @@ -14,6 +21,21 @@ import { TbankAccessService } from './tbank-access.service'; import { TbankClient } from './tbank.client'; import { DateOnlyRange, dateOnlyBounds } from './income-period'; +type OperationStream = { + account: AccountReferenceDto; + pending: BrokerageOperationDto[]; + cursor?: string; + finished: boolean; + seen: string[]; + lastOccurredAt?: number; +}; +type AllOperationsContinuation = { + signature: string; + range: { from: Date; to: Date }; + streams: OperationStream[]; + page?: Promise; +}; + const freshForMs = 30_000; const maxCachedPages = 100; @@ -25,6 +47,15 @@ export class BrokerageOperationsService { ttlAutopurge: true, }); + private readonly continuations = new LRUCache< + string, + AllOperationsContinuation + >({ + max: 100, + ttl: 5 * 60_000, + ttlAutopurge: true, + }); + constructor( private readonly access: TbankAccessService, private readonly tbank: TbankClient, @@ -45,6 +76,202 @@ export class BrokerageOperationsService { ); } + async getAll( + userId: string, + query: BrokerageOperationsQueryDto, + ): Promise { + return this.access.withSupportedAccounts( + userId, + async ({ token, generation, accounts }) => { + const signature = JSON.stringify({ + userId, + generation, + accounts: accounts.map((account) => account.id).sort(), + from: query.from ?? null, + to: query.to ?? null, + categories: [...new Set(query.categories ?? [])].sort(), + limit: query.limit ?? 100, + }); + const continuation = query.cursor + ? this.continuations.get(query.cursor) + : undefined; + if ( + query.cursor && + (!continuation || continuation.signature !== signature) + ) + throw new BadRequestException({ code: 'INVALID_OPERATIONS_CURSOR' }); + if (continuation?.page) return continuation.page; + const range = continuation?.range ?? this.range(query); + const streams: OperationStream[] = continuation + ? continuation.streams.map((stream) => ({ + ...stream, + pending: [...stream.pending], + seen: [...stream.seen], + })) + : accounts.map(({ id, name }) => ({ + account: { id, name }, + pending: [], + finished: false, + seen: [], + })); + const result = this.mergePage( + token, + generation, + userId, + query, + signature, + range, + streams, + ); + if (continuation) continuation.page = result; + try { + return await result; + } catch (error) { + if (continuation) continuation.page = undefined; + throw error; + } + }, + ); + } + + private async mergePage( + token: string, + generation: string, + userId: string, + query: BrokerageOperationsQueryDto, + signature: string, + range: { from: Date; to: Date }, + streams: OperationStream[], + ): Promise { + const items: BrokerageOperationDto[] = []; + // Each stream retains unconsumed rows. A page boundary never discards them. + while (items.length < (query.limit ?? 100)) { + for (const stream of streams) { + while (!stream.pending.length && !stream.finished) { + const page = await this.page( + token, + generation, + userId, + stream.account.id, + { ...query, cursor: stream.cursor }, + range, + ); + for (const item of page.items) { + const time = Date.parse(item.occurredAt); + if ( + !Number.isFinite(time) || + (stream.lastOccurredAt !== undefined && + time > stream.lastOccurredAt) + ) + throw new ServiceUnavailableException({ + code: 'TBANK_UNAVAILABLE', + }); + stream.lastOccurredAt = time; + } + if (page.hasNext) { + if (!page.nextCursor || stream.seen.includes(page.nextCursor)) + throw new ServiceUnavailableException({ + code: 'TBANK_UNAVAILABLE', + }); + stream.seen.push(page.nextCursor); + stream.cursor = page.nextCursor; + } else stream.finished = true; + stream.pending = page.items.map((item) => ({ + ...item, + account: stream.account, + })); + } + } + const next = streams + .filter((stream) => stream.pending.length) + .sort( + (left, right) => + Date.parse(right.pending[0].occurredAt) - + Date.parse(left.pending[0].occurredAt) || + left.account.id.localeCompare(right.account.id), + )[0]; + if (!next) break; + items.push(next.pending.shift()!); + } + const hasNext = streams.some( + (stream) => stream.pending.length || !stream.finished, + ); + const nextCursor = hasNext ? randomUUID() : null; + if (nextCursor) + this.continuations.set(nextCursor, { signature, range, streams }); + return { + accountId: 'all', + items, + hasNext, + nextCursor, + observedAt: new Date().toISOString(), + }; + } + + private async collectAllRanges( + userId: string, + ranges: + | DateOnlyRange[] + | ((accountOpenedAt: Date | undefined) => DateOnlyRange[]), + ): Promise<{ + accountOpenedAt: Date | undefined; + pages: BrokerageOperationsPageDto[][]; + }> { + return this.access.withSupportedAccounts( + userId, + async ({ token, generation, accounts }) => { + // The current cohort includes newly opened accounts with no earlier receipts. + const oldest = accounts + .map((account) => account.openedOn) + .filter((date): date is string => Boolean(date)) + .sort()[0]; + const accountOpenedAt = + oldest && accounts.every((account) => account.openedOn) + ? new Date(`${oldest}T00:00:00+03:00`) + : undefined; + const selected = + typeof ranges === 'function' ? ranges(accountOpenedAt) : ranges; + const pages: BrokerageOperationsPageDto[][] = selected.map(() => []); + for (const account of accounts) { + for (const [index, range] of selected.entries()) { + const collected = await this.collectRange( + token, + generation, + userId, + account.id, + range, + ); + pages[index].push( + ...collected.map((page) => ({ + ...page, + items: page.items.map((item) => ({ + ...item, + account: { id: account.id, name: account.name }, + })), + })), + ); + } + } + return { + accountOpenedAt, + pages: pages.map((group) => [ + { + ...emptyPage('all'), + items: group + .flatMap((page) => page.items) + .sort( + (left, right) => + Date.parse(right.occurredAt) - + Date.parse(left.occurredAt) || + left.account!.id.localeCompare(right.account!.id), + ), + }, + ]), + }; + }, + ); + } + async collectRanges( userId: string, accountId: string, @@ -55,6 +282,7 @@ export class BrokerageOperationsService { accountOpenedAt: Date | undefined; pages: BrokerageOperationsPageDto[][]; }> { + if (accountId === 'all') return this.collectAllRanges(userId, ranges); this.requestContext.set({ userId, brokerageAccountId: accountId }); return this.access.withAccount( userId, diff --git a/src/tbank/income-analytics.controller.ts b/src/tbank/income-analytics.controller.ts index e8744a5..c67d1f5 100644 --- a/src/tbank/income-analytics.controller.ts +++ b/src/tbank/income-analytics.controller.ts @@ -19,6 +19,21 @@ import { IncomeAnalyticsService } from './income-analytics.service'; export class IncomeAnalyticsController { constructor(private readonly income: IncomeAnalyticsService) {} + @Get('all/income-summary') + @ApiOkResponse({ type: IncomeSummaryDto }) + allSummary(@Req() request: Request & { user: AccessPayload }) { + return this.income.summary(request.user.sub, 'all'); + } + + @Get('all/income-analytics') + @ApiOkResponse({ type: IncomeAnalyticsDto }) + allAnalytics( + @Req() request: Request & { user: AccessPayload }, + @Query() query: IncomeAnalyticsQueryDto, + ) { + return this.income.analytics(request.user.sub, 'all', query.from, query.to); + } + @Get(':accountId/income-summary') @ApiOkResponse({ type: IncomeSummaryDto }) summary( diff --git a/src/tbank/income-analytics.service.ts b/src/tbank/income-analytics.service.ts index 2713afe..e8f51eb 100644 --- a/src/tbank/income-analytics.service.ts +++ b/src/tbank/income-analytics.service.ts @@ -228,7 +228,7 @@ function instrumentKey(entry: BrokerageOperationDto): string { if (entry.figi) return `figi:${entry.figi}`; if (entry.ticker || entry.instrumentType) return `ticker:${entry.ticker ?? ''}:${entry.instrumentType ?? ''}`; - return `operation:${entry.id}`; + return `operation:${entry.account ? `${entry.account.id}:` : ''}${entry.id}`; } function difference( diff --git a/src/tbank/portfolio-totals.ts b/src/tbank/portfolio-totals.ts index 4bdfe86..a507e48 100644 --- a/src/tbank/portfolio-totals.ts +++ b/src/tbank/portfolio-totals.ts @@ -1,4 +1,11 @@ -import { add, decimal, formatDecimal, isZero, multiply } from './decimal'; +import { + add, + decimal, + decimalFromString, + formatDecimal, + isZero, + multiply, +} from './decimal'; import { MoneyDto } from './portfolio.dto'; import { TbankPortfolio, TbankPositions } from './tbank.client'; @@ -34,4 +41,20 @@ function totalsByCurrency( .map(([currency, value]) => ({ currency, value: formatDecimal(value) })); } -export { totalsByCurrency }; +function sumMoneyByCurrency(amounts: MoneyDto[]): MoneyDto[] { + const totals = new Map>(); + for (const amount of amounts) { + totals.set( + amount.currency, + add( + totals.get(amount.currency) ?? decimal({}), + decimalFromString(amount.value), + ), + ); + } + return [...totals] + .sort(([left], [right]) => left.localeCompare(right)) + .map(([currency, value]) => ({ currency, value: formatDecimal(value) })); +} + +export { totalsByCurrency, sumMoneyByCurrency }; diff --git a/src/tbank/portfolio.controller.ts b/src/tbank/portfolio.controller.ts index 6dc3d22..c1a9be9 100644 --- a/src/tbank/portfolio.controller.ts +++ b/src/tbank/portfolio.controller.ts @@ -15,6 +15,12 @@ import { PortfolioService } from './portfolio.service'; export class PortfolioController { constructor(private readonly portfolio: PortfolioService) {} + @Get('all/portfolio') + @ApiOkResponse({ type: PortfolioDto }) + all(@Req() request: Request & { user: AccessPayload }) { + return this.portfolio.getAll(request.user.sub); + } + @Get(':accountId/portfolio') @ApiOkResponse({ type: PortfolioDto }) get( diff --git a/src/tbank/portfolio.dto.ts b/src/tbank/portfolio.dto.ts index 7549b3d..2f6f10b 100644 --- a/src/tbank/portfolio.dto.ts +++ b/src/tbank/portfolio.dto.ts @@ -1,4 +1,5 @@ -import { ApiProperty } from '@nestjs/swagger'; +import { ApiProperty, ApiPropertyOptional } from '@nestjs/swagger'; +import { AccountReferenceDto } from './account-reference.dto'; export class MoneyDto { @ApiProperty() @@ -9,6 +10,9 @@ export class MoneyDto { } export class PortfolioPositionDto { + @ApiPropertyOptional({ type: AccountReferenceDto }) + account?: AccountReferenceDto; + @ApiProperty({ type: String, nullable: true }) figi: string | null; diff --git a/src/tbank/portfolio.service.ts b/src/tbank/portfolio.service.ts index 25c4d69..e5b5e1e 100644 --- a/src/tbank/portfolio.service.ts +++ b/src/tbank/portfolio.service.ts @@ -8,9 +8,12 @@ import { multiply, subtract, } from './decimal'; -import { totalsByCurrency } from './portfolio-totals'; +import { totalsByCurrency, sumMoneyByCurrency } from './portfolio-totals'; import { MoneyDto, PortfolioDto, PortfolioPositionDto } from './portfolio.dto'; -import { TbankAccessService } from './tbank-access.service'; +import { + TbankAccessService, + TbankConnectionContext, +} from './tbank-access.service'; import { InstrumentMetadataService } from './instrument-metadata.service'; import { TbankClient, TbankMoney } from './tbank.client'; @@ -46,49 +49,86 @@ export class PortfolioService { async get(userId: string, accountId: string): Promise { this.requestContext.set({ userId, brokerageAccountId: accountId }); - const key = `${userId}:${accountId}`; - return this.access.withConnection(userId, async ({ token, generation }) => { - const cached = this.snapshots.get(key); - if ( - cached?.generation === generation && - Date.now() - cached.storedAt <= freshForMs - ) - return cached.snapshot; - try { - const snapshot = await this.access.withAccountForConnection( - { token, generation }, - accountId, - async () => { - const [portfolio, positions] = await Promise.all([ - this.tbank.getPortfolio(token, accountId), - this.tbank.getPositions(token, accountId), - ]); - return this.normalize( - token, - portfolio.positions ?? [], - positions.money ?? [], - ); - }, - ); - this.snapshots.set(key, { - generation, - snapshot, - storedAt: Date.now(), + return this.access.withConnection(userId, (connection) => + this.snapshot(userId, accountId, connection), + ); + } + + async getAll(userId: string): Promise { + return this.access.withSupportedAccounts(userId, async (context) => { + const portfolios: PortfolioDto[] = []; + for (const account of context.accounts) { + const portfolio = await this.snapshot(userId, account.id, context); + portfolios.push({ + ...portfolio, + positions: portfolio.positions.map((position) => ({ + ...position, + account: { id: account.id, name: account.name }, + })), }); - return snapshot; - } catch (error) { - if ( - error instanceof ServiceUnavailableException && - cached?.generation === generation && - Date.now() - cached.storedAt <= fallbackForMs - ) { - return { ...cached.snapshot, isStale: true }; - } - throw error; } + return { + lastSuccessfulAt: + portfolios.map((portfolio) => portfolio.lastSuccessfulAt).sort()[0] ?? + new Date().toISOString(), + isStale: portfolios.some((portfolio) => portfolio.isStale), + totalsByCurrency: sumMoneyByCurrency( + portfolios.flatMap((portfolio) => portfolio.totalsByCurrency), + ), + cashBalances: sumMoneyByCurrency( + portfolios.flatMap((portfolio) => portfolio.cashBalances), + ), + positions: portfolios.flatMap((portfolio) => portfolio.positions), + }; }); } + private async snapshot( + userId: string, + accountId: string, + { token, generation }: TbankConnectionContext, + ): Promise { + const key = `${userId}:${accountId}`; + const cached = this.snapshots.get(key); + if ( + cached?.generation === generation && + Date.now() - cached.storedAt <= freshForMs + ) + return cached.snapshot; + try { + const snapshot = await this.access.withAccountForConnection( + { token, generation }, + accountId, + async () => { + const [portfolio, positions] = await Promise.all([ + this.tbank.getPortfolio(token, accountId), + this.tbank.getPositions(token, accountId), + ]); + return this.normalize( + token, + portfolio.positions ?? [], + positions.money ?? [], + ); + }, + ); + this.snapshots.set(key, { + generation, + snapshot, + storedAt: Date.now(), + }); + return snapshot; + } catch (error) { + if ( + error instanceof ServiceUnavailableException && + cached?.generation === generation && + Date.now() - cached.storedAt <= fallbackForMs + ) { + return { ...cached.snapshot, isStale: true }; + } + throw error; + } + } + invalidate(userId: string): void { for (const key of this.snapshots.keys()) if (key.startsWith(`${userId}:`)) this.snapshots.delete(key); diff --git a/src/tbank/upcoming-events.controller.ts b/src/tbank/upcoming-events.controller.ts index 6c3a95b..3237d71 100644 --- a/src/tbank/upcoming-events.controller.ts +++ b/src/tbank/upcoming-events.controller.ts @@ -15,6 +15,12 @@ import { UpcomingEventsService } from './upcoming-events.service'; export class UpcomingEventsController { constructor(private readonly events: UpcomingEventsService) {} + @Get('all/events') + @ApiOkResponse({ type: UpcomingEventsDto }) + all(@Req() request: Request & { user: AccessPayload }) { + return this.events.getAll(request.user.sub); + } + @Get(':accountId/events') @ApiOkResponse({ type: UpcomingEventsDto }) get( diff --git a/src/tbank/upcoming-events.dto.ts b/src/tbank/upcoming-events.dto.ts index c1ec593..7c9faef 100644 --- a/src/tbank/upcoming-events.dto.ts +++ b/src/tbank/upcoming-events.dto.ts @@ -1,7 +1,11 @@ -import { ApiProperty } from '@nestjs/swagger'; +import { ApiProperty, ApiPropertyOptional } from '@nestjs/swagger'; +import { AccountReferenceDto } from './account-reference.dto'; import { MoneyDto } from './portfolio.dto'; export class UpcomingEventDto { + @ApiPropertyOptional({ type: AccountReferenceDto }) + account?: AccountReferenceDto; + @ApiProperty({ enum: ['coupon', 'dividend', 'maturity', 'offer'] }) type: 'coupon' | 'dividend' | 'maturity' | 'offer'; diff --git a/src/tbank/upcoming-events.service.ts b/src/tbank/upcoming-events.service.ts index ab27441..a347425 100644 --- a/src/tbank/upcoming-events.service.ts +++ b/src/tbank/upcoming-events.service.ts @@ -48,46 +48,80 @@ export class UpcomingEventsService { async get(userId: string, accountId: string): Promise { this.requestContext.set({ userId, brokerageAccountId: accountId }); - return this.access.withAccount(userId, accountId, async ({ token }) => { - const portfolio = await this.tbank.getPortfolio(token, accountId); - const window = dateWindow(); - const supportedPositions = (portfolio.positions ?? []).filter(supported); - const events: UpcomingEventDto[][] = []; - for ( - let offset = 0; - offset < supportedPositions.length; - offset += positionsBatchSize - ) { - const batch = supportedPositions.slice( - offset, - offset + positionsBatchSize, - ); - events.push( - ...(await Promise.all( - batch.map((position) => this.positionEvents(token, position)), - )), - ); - } - const projected = events - .flat() - .filter( - (event) => event.date >= window.from && event.date <= window.to, - ); - this.logger.info('upcoming-events.calculated', { - count: projected.length, - }); - return { - accountId, - calculatedAt: new Date().toISOString(), - events: projected.sort( + return this.access.withAccount(userId, accountId, ({ token }) => + this.forAccount(token, accountId), + ); + } + + async getAll(userId: string): Promise { + return this.access.withSupportedAccounts( + userId, + async ({ token, accounts }) => { + const events: UpcomingEventDto[] = []; + for (const account of accounts) { + const result = await this.forAccount(token, account.id); + events.push( + ...result.events.map((event) => ({ + ...event, + account: { id: account.id, name: account.name }, + })), + ); + } + events.sort( (left, right) => left.date.localeCompare(right.date) || - left.type.localeCompare(right.type) || - left.ticker.localeCompare(right.ticker), - ), - incomeTotalsByCurrency: incomeTotals(projected), - }; + left.account!.id.localeCompare(right.account!.id), + ); + return { + accountId: 'all', + calculatedAt: new Date().toISOString(), + events, + incomeTotalsByCurrency: incomeTotals(events), + }; + }, + ); + } + + private async forAccount( + token: string, + accountId: string, + ): Promise { + const portfolio = await this.tbank.getPortfolio(token, accountId); + const window = dateWindow(); + const supportedPositions = (portfolio.positions ?? []).filter(supported); + const events: UpcomingEventDto[][] = []; + for ( + let offset = 0; + offset < supportedPositions.length; + offset += positionsBatchSize + ) { + const batch = supportedPositions.slice( + offset, + offset + positionsBatchSize, + ); + events.push( + ...(await Promise.all( + batch.map((position) => this.positionEvents(token, position)), + )), + ); + } + const projected = events + .flat() + .filter((event) => event.date >= window.from && event.date <= window.to); + this.logger.info('upcoming-events.calculated', { + count: projected.length, }); + return { + accountId, + calculatedAt: new Date().toISOString(), + events: projected.sort( + (left, right) => + left.date.localeCompare(right.date) || + left.type.localeCompare(right.type) || + left.ticker.localeCompare(right.ticker), + ), + incomeTotalsByCurrency: incomeTotals(projected), + }; } private async positionEvents( diff --git a/test/all-accounts.e2e-spec.ts b/test/all-accounts.e2e-spec.ts new file mode 100644 index 0000000..a87ff05 --- /dev/null +++ b/test/all-accounts.e2e-spec.ts @@ -0,0 +1,101 @@ +import { ConflictException, INestApplication } from '@nestjs/common'; +import { JwtService } from '@nestjs/jwt'; +import { Test } from '@nestjs/testing'; +import request from 'supertest'; +import type { Server } from 'node:http'; +import { AppModule } from '../src/app.module'; +import { configureApp } from '../src/main'; +import { TbankAccessService } from '../src/tbank/tbank-access.service'; + +describe('All accounts HTTP boundary', () => { + let app: INestApplication; + const withSupportedAccounts = jest.fn(); + const token = new JwtService().sign( + { sub: 'account-owner', type: 'access' }, + { secret: process.env.JWT_SECRET }, + ); + const paths = [ + 'portfolio', + 'operations', + 'income-summary', + 'income-analytics', + 'events', + ]; + + beforeAll(async () => { + const module = await Test.createTestingModule({ imports: [AppModule] }) + .overrideProvider(TbankAccessService) + .useValue({ withSupportedAccounts }) + .compile(); + app = module.createNestApplication(); + configureApp(app); + await app.init(); + }); + beforeEach(() => { + withSupportedAccounts.mockReset(); + withSupportedAccounts.mockImplementation( + (_user: string, handler: (context: unknown) => unknown) => + handler({ token: 'unused', generation: 'first', accounts: [] }), + ); + }); + afterAll(() => app.close()); + + it.each(paths)('requires authentication for /all/%s', async (path) => { + await request(app.getHttpServer()) + .get(`/api/v1/brokerage-accounts/all/${path}`) + .expect(401); + expect(withSupportedAccounts).not.toHaveBeenCalled(); + }); + + it.each(paths)( + 'routes /all/%s to the aggregate with verified identity', + async (path) => { + const response = await request(app.getHttpServer()) + .get(`/api/v1/brokerage-accounts/all/${path}`) + .set('Authorization', `Bearer ${token}`) + .expect(200); + expect(withSupportedAccounts).toHaveBeenCalledWith( + 'account-owner', + expect.any(Function), + ); + expect(response.body).not.toHaveProperty('token'); + expect(response.body).not.toHaveProperty('generation'); + if (path === 'portfolio') + expect(response.body).toMatchObject({ positions: [] }); + else expect(response.body).toMatchObject({ accountId: 'all' }); + }, + ); + + it.each([ + 'operations?limit=0', + 'operations?limit=1001', + 'operations?categories=unknown', + 'operations?from=not-a-date', + 'income-analytics?from=2026-08-01', + 'income-analytics?from=2026-08-31&to=2026-08-01', + ])('validates /all/%s in the configured pipeline', async (path) => { + await request(app.getHttpServer()) + .get(`/api/v1/brokerage-accounts/all/${path}`) + .set('Authorization', `Bearer ${token}`) + .expect(400); + }); + + it('returns a recoverable code for an expired or foreign continuation', async () => { + const response = await request(app.getHttpServer()) + .get('/api/v1/brokerage-accounts/all/operations?cursor=unknown') + .set('Authorization', `Bearer ${token}`) + .expect(400); + expect(response.body).toMatchObject({ code: 'INVALID_OPERATIONS_CURSOR' }); + }); + + it('preserves connection errors without a partial success response', async () => { + withSupportedAccounts.mockRejectedValue( + new ConflictException({ code: 'TBANK_NOT_CONNECTED' }), + ); + const response = await request(app.getHttpServer()) + .get('/api/v1/brokerage-accounts/all/portfolio') + .set('Authorization', `Bearer ${token}`) + .expect(409); + expect(response.body).toMatchObject({ code: 'TBANK_NOT_CONNECTED' }); + }); +});