From 9717565bc399167d3c03159e4198d7e21d226c81 Mon Sep 17 00:00:00 2001 From: Sergey Krylov Date: Thu, 15 Oct 2026 07:37:17 +0300 Subject: [PATCH] refactor: remove unreachable catalog producer and sync orchestration (WealthLedger/backend#62) --- eslint-suppressions.json | 8 +- .../application/catalog-discovery.service.ts | 206 +----------- .../investments/tbank-sync/tbank-sync.dto.ts | 27 -- .../tbank-sync/tbank-sync.service.ts | 293 +--------------- test/tbank-catalog-discovery.e2e-spec.ts | 316 ++++-------------- 5 files changed, 75 insertions(+), 775 deletions(-) diff --git a/eslint-suppressions.json b/eslint-suppressions.json index aa4eafc..4506661 100644 --- a/eslint-suppressions.json +++ b/eslint-suppressions.json @@ -54,9 +54,6 @@ } }, "src/modules/investments/instruments/application/catalog-discovery.service.ts": { - "sonarjs/cognitive-complexity": { - "count": 1 - }, "sonarjs/no-nested-conditional": { "count": 1 } @@ -142,11 +139,8 @@ } }, "src/modules/investments/tbank-sync/tbank-sync.service.ts": { - "sonarjs/cognitive-complexity": { - "count": 1 - }, "sonarjs/no-nested-conditional": { - "count": 7 + "count": 6 } }, "test/bank-cards-durability.cjs": { diff --git a/src/modules/investments/instruments/application/catalog-discovery.service.ts b/src/modules/investments/instruments/application/catalog-discovery.service.ts index f1c62c4..8202b42 100644 --- a/src/modules/investments/instruments/application/catalog-discovery.service.ts +++ b/src/modules/investments/instruments/application/catalog-discovery.service.ts @@ -3,15 +3,11 @@ 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 { TbankAccessService, TbankClient, tbankSourceTtlMs, - type TbankSyncAttempt, } from '../../../../integrations/tbank/tbank.api'; import { investmentResourceMeta } from '../../investment-resource.mapper'; import { InstrumentErrorCode } from '../domain/instrument-error-code'; @@ -31,42 +27,22 @@ import { publicInstrument } from './public-instrument'; const UNAMBIGUOUS_MATCH_COUNT = 1; const NO_MATCHES = 0; const FIRST_RESULT_OFFSET = 0; -const MISSING_DATA_REVISION = '0'; const tickerAmbiguityLimit = 2; -type Producer = { - attempt: TbankSyncAttempt; - force: boolean; - input: CatalogQuery; - observers: TbankSyncAttempt[]; -}; @Injectable() export class CatalogDiscoveryService { - private readonly runId = randomUUID(); - private readonly active = new Set(); - private readonly producers = new Map(); - private readonly queue: Producer[] = []; - private reservations: Promise = Promise.resolve(); - private running = false; - private readonly completions = new Map< - string, - { done: Promise; resolve: () => void; workId: string } - >(); - constructor( private readonly repository: CatalogDiscoveryRepository, private readonly access: TbankAccessService, private readonly tbank: TbankClient, - private readonly logger: StructuredLogger, private readonly enrichment: CatalogEnrichmentService, ) {} - async read(userId: string, input: CatalogQuery, observe = false) { + async read(userId: string, input: CatalogQuery) { input = catalogQuery(input); const connection = await this.access.local(userId); const saved = await this.repository.read(input); - if (!observe) await this.request(userId, connection.revision, input); return { data: saved.data, meta: investmentResourceMeta({ @@ -82,7 +58,7 @@ export class CatalogDiscoveryService { async search(userId: string, input: CatalogQuery, observe = false) { if (!observe) return this.onlineSearch(userId, catalogQuery(input)); - const discovery = await this.read(userId, input, observe); + const discovery = await this.read(userId, input); const sources = discovery.data ? await this.enrichment.readMany( userId, @@ -182,11 +158,11 @@ export class CatalogDiscoveryService { instrument?.ticker && /^[A-Z0-9][\w.@+-]{0,79}$/i.test(instrument.ticker) ) { - const alias = await this.read( - userId, - { instrumentType, query: instrument.ticker, exactTicker: true }, - observe, - ); + const alias = await this.read(userId, { + instrumentType, + query: instrument.ticker, + exactTicker: true, + }); if (alias.data?.length === UNAMBIGUOUS_MATCH_COUNT && alias.data[0].uid === uid) { slug = instrument.ticker.toUpperCase(); } @@ -247,11 +223,11 @@ export class CatalogDiscoveryService { const result = await this.detail(userId, instrumentType, matches[0].uid); return { ...result, data: result.data ? { ...result.data, slug: query } : null }; } - const lookup = await this.read( - userId, - { instrumentType, query: ticker, exactTicker: true }, - observe, - ); + const lookup = await this.read(userId, { + instrumentType, + query: ticker, + exactTicker: true, + }); if (lookup.data?.length === NO_MATCHES) throw new NotFoundException(); if (lookup.data && lookup.data.length > UNAMBIGUOUS_MATCH_COUNT) { throw new ConflictException({ @@ -337,162 +313,4 @@ export class CatalogDiscoveryService { }), }; } - - request( - userId: string, - revision: string, - query: CatalogQuery, - force = false, - ) { - const input = catalogQuery(query); - const reserved = this.reservations.then(async () => { - const saved = await this.repository.read(input); - if ( - !force && - saved.dataRevision !== MISSING_DATA_REVISION - ) return null; - const reservation = await this.repository.reserve( - { userId, revision }, - input, - this.runId, - this.active, - ); - if (!reservation) return null; - const { id, attemptId } = reservation.work; - if (!reservation.launch) { - const current = this.producers.get(catalogScope(input)); - if (current) current.force ||= force; - return id; - } - const attempt = { userId, revision, id, attemptId }; - this.active.add(attemptId); - let resolve = () => {}; - const done = new Promise((finish) => { - resolve = finish; - }); - this.completions.set(attemptId, { workId: id, done, resolve }); - const key = catalogScope(input); - const producer = this.producers.get(key); - if (producer) { - producer.observers.push(attempt); - producer.force ||= force; - } else { - const queued = { input, attempt, observers: [], force }; - this.producers.set(key, queued); - this.queue.push(queued); - } - if (!this.running) { - this.running = true; - setImmediate(() => { - void this.drain(); - }); - } - return id; - }); - this.reservations = reserved.catch(() => undefined); - return reserved; - } - - async refresh(userId: string, revision: string, query: CatalogQuery) { - const id = await this.request(userId, revision, query, true); - if (id) { - await Promise.all( - [...this.completions.values()] - .filter((entry) => entry.workId === id) - .map((entry) => entry.done), - ); - } - const saved = await this.repository.read(catalogQuery(query)); - const sources = await this.enrichment.readMany( - userId, - query.instrumentType, - saved.data?.map((item) => item.uid) ?? [], - true, - ); - return [ - ...(id ? [id] : []), - ...sources.flatMap((source) => - source.meta.sync.id ? [source.meta.sync.id] : [], - ), - ]; - } - - private async drain() { - try { - while (this.queue.length) { - const producer = this.queue.shift(); - if (!producer) break; - const { input, attempt } = producer; - let success = false; - try { - const token = await this.access.tokenForAttempt(attempt); - if (token) { - const found = - input.instrumentType === 'share' - ? await this.tbank.findShares(token, input.query) - : await this.tbank.findBonds(token, input.query); - success = await this.repository.publish( - attempt, - input, - selectCatalogInstruments( - input.exactTicker - ? found.filter( - (item) => - item.ticker?.trim().toUpperCase() === input.query, - ) - : found, - input.query, - input.instrumentType, - ).slice( - FIRST_RESULT_OFFSET, - input.exactTicker ? tickerAmbiguityLimit : catalogSearchLimit, - ), - new Date(), - ); - } - } catch (error) { - if (error instanceof Error && isTbankErrorCode(error.message, TbankErrorCode.InvalidToken)) { - await this.access.rejectMembershipToken(attempt); - } else { - this.logger.event('warn', 'catalog-discovery.failed', { - code: TbankErrorCode.Unavailable, - }); - } - } - if (success) { - const saved = await this.repository.read(input); - for (const observer of [attempt, ...producer.observers]) { - if (!input.exactTicker || saved.data?.length === UNAMBIGUOUS_MATCH_COUNT) { - await this.enrichment.requestMany( - observer.userId, - observer.revision, - input.instrumentType, - saved.data?.map((item) => item.uid) ?? [], - producer.force, - input.exactTicker ? ['detail'] : ['last-price', 'detail'], - ); - } - } - } - this.producers.delete(catalogScope(input)); - for (const observer of [attempt, ...producer.observers]) { - await this.repository.finish(observer, success); - this.active.delete(observer.attemptId); - this.completions.get(observer.attemptId)?.resolve(); - this.completions.delete(observer.attemptId); - } - } - } catch { - for (const completion of this.completions.values()) completion.resolve(); - this.completions.clear(); - this.active.clear(); - this.producers.clear(); - this.queue.length = 0; - this.logger.event('error', 'catalog-discovery.persistence-failed', { - code: TbankErrorCode.Unavailable, - }); - } finally { - this.running = false; - } - } } diff --git a/src/modules/investments/tbank-sync/tbank-sync.dto.ts b/src/modules/investments/tbank-sync/tbank-sync.dto.ts index 2eb0741..cfb5d78 100644 --- a/src/modules/investments/tbank-sync/tbank-sync.dto.ts +++ b/src/modules/investments/tbank-sync/tbank-sync.dto.ts @@ -209,13 +209,6 @@ export const investmentScreenSchema = { }, }; -export class TbankSyncAcceptedDto { - @ApiProperty({ format: 'uuid' }) declare id: string; - @ApiProperty({ type: Object, ...investmentScreenSchema }) - declare scope: InvestmentScreenDto; - - @ApiProperty({ format: 'date-time' }) declare now: string; -} export class TbankSyncStatusDto { @ApiProperty({ format: 'uuid' }) declare id: string; @ApiProperty({ type: Object, ...investmentScreenSchema, nullable: true }) @@ -229,26 +222,6 @@ export class TbankSyncStatusDto { null | string; } -export class HourlySyncSectionDto { - @ApiProperty({ enum: ['membership', 'portfolio', 'operations', 'events'] }) - declare section: 'events' | 'membership' | 'operations' | 'portfolio'; - - @ApiPropertyOptional({ - type: String, - nullable: false, - description: 'Account identifier of the problematic section, when scoped.', - }) - declare accountId?: string; - - @ApiProperty({ - type: String, - nullable: true, - description: - 'Safe error code of the failed section. No tokens or upstream payloads.', - }) - declare errorCode: null | string; -} - export class HourlySyncStatusDto { @ApiProperty({ enum: ['running', 'succeeded', 'partial', 'failed', 'cancelled'], diff --git a/src/modules/investments/tbank-sync/tbank-sync.service.ts b/src/modules/investments/tbank-sync/tbank-sync.service.ts index 7c59e86..bdc37a8 100644 --- a/src/modules/investments/tbank-sync/tbank-sync.service.ts +++ b/src/modules/investments/tbank-sync/tbank-sync.service.ts @@ -2,314 +2,27 @@ import { ConflictException, Injectable, NotFoundException, - ServiceUnavailableException, } from '@nestjs/common'; import { randomUUID } from 'node:crypto'; -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 { - CandleService, - CatalogDiscoveryService, - CatalogEnrichmentService, -} from '../instruments/instruments.api'; -import { planInvestmentScreen } from '../investment-sync-plan'; -import { - BrokerageOperationsSyncService, - type OperationInterval, -} from '../operations/operations.api'; -import { PortfolioService } from '../portfolio/portfolio.api'; -import { normalizeScreen } from './tbank-screen'; -import { InvestmentScreenDto, TbankSyncStatusDto } from './tbank-sync.dto'; +import { TbankAccessService } from '../../../integrations/tbank/tbank.api'; +import { BrokerageOperationsSyncService } from '../operations/operations.api'; +import { TbankSyncStatusDto } from './tbank-sync.dto'; import { type ScreenIntent, TbankSyncRepository, } from './tbank-sync.repository'; -const FOREGROUND_CANDLE_PRIORITY = 0; - @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 metadata: InstrumentMetadataService, - private readonly discovery: CatalogDiscoveryService, - private readonly sources: CatalogEnrichmentService, - private readonly candles: CandleService, ) {} - 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, - ); - 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) diff --git a/test/tbank-catalog-discovery.e2e-spec.ts b/test/tbank-catalog-discovery.e2e-spec.ts index cab7432..3dc6f2b 100644 --- a/test/tbank-catalog-discovery.e2e-spec.ts +++ b/test/tbank-catalog-discovery.e2e-spec.ts @@ -1,6 +1,5 @@ import { ConfigService } from '@nestjs/config'; import assert from 'node:assert/strict'; -import { setTimeout as delay } from 'node:timers/promises'; import { TbankAccessService } from '../src/integrations/tbank/tbank-access.service'; import { TbankTokenService } from '../src/integrations/tbank/tbank-token.service'; @@ -12,7 +11,7 @@ import { } from '../src/modules/investments/instruments/infrastructure/catalog-discovery.repository'; import { PrismaService } from '../src/prisma/prisma.service'; -// Real transactions prove query coverage, private work ownership and late-publication fences. +// Real transactions prove query normalization, confirmed-empty identity and cross-user isolation of shared reads. describe('shared catalog discovery (SQLite)', () => { const prisma = new PrismaService( new ConfigService({ DATABASE_URL: process.env.DATABASE_URL }), @@ -30,7 +29,6 @@ describe('shared catalog discovery (SQLite)', () => { repository, access, upstream as never, - { event: jest.fn() } as never, { requestMany: jest.fn().mockResolvedValue([]) } as never, ); let service: CatalogDiscoveryService; @@ -51,14 +49,42 @@ describe('shared catalog discovery (SQLite)', () => { connectedAt: new Date(), }, }); - async function settled() { - for (let i = 0; i < 300; i++) { - if (!(await prisma.tbankSyncWork.count({ where: { state: 'running' } }))) { - return; - } - await delay(5); + // Writes a finished discovery directly, so the read path is tested without any producer. + async function seedCatalog( + query: string, + instruments: Array<{ payload: CatalogInstrument; uid: string }>, + ) { + for (const item of instruments) { + await prisma.tbankInstrument.create({ + data: { + provider: 'tbank', + uid: item.uid, + instrumentType: 'share', + metadata: {}, + fetchedAt: new Date(), + }, + }); + await prisma.tbankInstrumentSource.create({ + data: { + provider: 'tbank', + instrumentUid: item.uid, + source: 'tbank', + kind: 'catalog', + rangeKey: 'share', + payload: item.payload, + fetchedAt: new Date(), + }, + }); } - throw new Error('Catalog work did not settle'); + await prisma.tbankCatalogDiscovery.create({ + data: { + provider: 'tbank', + instrumentType: 'share', + query, + instrumentUids: instruments.map((item) => item.uid), + fetchedAt: new Date(), + }, + }); } beforeAll(async () => { await prisma.$connect(); @@ -72,265 +98,41 @@ describe('shared catalog discovery (SQLite)', () => { beforeEach(async () => { await prisma.tbankConnection.deleteMany(); await prisma.tbankCatalogDiscovery.deleteMany(); + await prisma.tbankInstrumentSource.deleteMany(); await prisma.tbankInstrument.deleteMany(); await seed('owner'); await seed('other'); jest.resetAllMocks(); - upstream.findShares.mockResolvedValue([instrument]); - upstream.findBonds.mockResolvedValue([ - { ...instrument, uid: 'bond', instrumentKind: 'INSTRUMENT_TYPE_BOND' }, - ]); service = createService(); }); - it('persists normalized query identity and confirmed empty separately from unknown, across users and reopen', async () => { + it('treats an unknown query as unavailable and a seeded query as available for another user', async () => { const cold = await service.read('owner', { ...query, query: ' sber ' }); expect(cold.data).toBeNull(); - expect(cold.meta.availability).toBe('initial-loading'); - await settled(); - await prisma.$disconnect(); - service = createService(); - const decrypt = jest.spyOn(tokens, 'decrypt'); + expect(cold.meta.availability).toBe('temporarily-unavailable'); + await seedCatalog('SBER', [{ uid: 'uid', payload: instrument }]); const warm = await service.read('other', query); expect(warm.data).toEqual([instrument]); expect(warm.meta.availability).toBe('available'); - expect(warm.meta.completeness).toBe('complete'); - expect(warm.meta.sync.id).toBeNull(); - expect(decrypt).not.toHaveBeenCalled(); - expect(upstream.findShares).toHaveBeenCalledTimes(1); - expect( - (await service.read('other', { ...query, query: 'GAZP' }, true)).data, - ).toBeNull(); - upstream.findShares.mockResolvedValue([]); - await service.read('other', { ...query, query: 'GAZP' }); - await settled(); - expect( - (await service.read('owner', { ...query, query: 'GAZP' })).data, - ).toEqual([]); - expect((await service.read('owner', query)).data).toEqual([instrument]); - expect(upstream.findShares).toHaveBeenCalledTimes(2); - decrypt.mockRestore(); - await service.read('owner', { ...query, instrumentType: 'bond' }); - await settled(); - expect( - (await service.read('other', { ...query, instrumentType: 'bond' })) - .data?.[0].uid, - ).toBe('bond'); - }); - it('keeps selection ordering, deduplicates before limiting, and retains identity without enrichment or private fields', async () => { - upstream.findShares.mockResolvedValue([ - { ...instrument, uid: 'unavailable', apiTradeAvailableFlag: false }, - { - ...instrument, - uid: 'preferred', - apiTradeAvailableFlag: true, - classCode: 'TQBR', - quantity: '999', - estimatedAmount: 'secret', - }, - ...Array.from({ length: 40 }, (_, i) => ({ - ...instrument, - uid: `other-${String(i).padStart(2, '0')}`, - isin: undefined, - ticker: 'OTHER', - })), - { - ...instrument, - uid: 'wrong-kind', - instrumentKind: 'INSTRUMENT_TYPE_BOND', - }, - ]); - await service.read('owner', query); - await settled(); - const saved = await service.read('other', query); - expect(saved.data).toHaveLength(30); - expect(saved.data?.[0].uid).toBe('preferred'); - expect( - saved.data?.some((row) => - ['unavailable', 'wrong-kind'].includes(row.uid), - ), - ).toBe(false); - expect( - JSON.stringify(await prisma.tbankInstrumentSource.findMany()), - ).not.toMatch( - /quantity|estimatedAmount|secret|owner-revision|synthetic-owner/, - ); - expect(await prisma.tbankInstrumentLookup.count()).toBe(0); - }); - it('does not overwrite detailed metadata or its timestamp while refreshing discovery', async () => { - const detailed = { ...instrument, sector: 'financial', lot: 10 }; - const fetchedAt = new Date('2026-01-01'); - await prisma.tbankInstrument.create({ - data: { - provider: 'tbank', - uid: 'uid', - instrumentType: 'share', - metadata: detailed, - fetchedAt, - }, - }); - await service.read('owner', query); - await settled(); - const row = await prisma.tbankInstrument.findUniqueOrThrow({ - where: { provider_uid: { provider: 'tbank', uid: 'uid' } }, - }); - expect(row.metadata).toEqual(detailed); - expect(row.fetchedAt).toEqual(fetchedAt); - }); - it('coalesces twenty concurrent requests and forces with private handles and one upstream call', async () => { - const blocked = deferred(); - upstream.findShares.mockReturnValue(blocked.promise); - const reads = await Promise.all( - Array.from({ length: 20 }, (_, i) => - service.read(i % 2 ? 'owner' : 'other', query), - ), - ); - await service.request('owner', 'owner-revision', query, true); - await delay(10); - expect(upstream.findShares).toHaveBeenCalledTimes(1); - expect(new Set(reads.map((read) => read.meta.sync.id)).size).toBe(2); - blocked.resolve([instrument]); - await settled(); - expect( - await prisma.tbankSyncWork.count({ where: { state: 'complete' } }), - ).toBe(2); - }); - it.each(['delete', 'replace'])( - 'discards late %s publication and does not borrow observer credentials', - async (action) => { - const blocked = deferred(); - upstream.findShares.mockReturnValueOnce(blocked.promise); - await service.read('owner', query); - await service.read('other', query); - await delay(10); - await prisma.tbankConnection.delete({ where: { userId: 'owner' } }); - if (action === 'replace') { - await prisma.tbankConnection.create({ - data: { - userId: 'owner', - revision: 'replacement', - encryptedToken: tokens.encrypt('replacement'), - connectedAt: new Date(), - }, - }); - } - blocked.resolve([instrument]); - await settled(); - expect(await prisma.tbankCatalogDiscovery.count()).toBe(0); - expect(await prisma.tbankInstrument.count()).toBe(0); - expect( - (await service.read('other', query, true)).meta.sync.errorCode, - ).toBe('TBANK_UNAVAILABLE'); - expect(upstream.findShares).toHaveBeenCalledTimes(1); - await service.read('other', query); - await settled(); - expect((await service.read('other', query, true)).data).toEqual([ - instrument, - ]); - }, - ); - it('preserves a successful result and its timestamp after a failed force without observe retry', async () => { - await service.read('owner', query); - await settled(); - const before = await repository.read(query); - upstream.findShares.mockRejectedValue(new Error('outage')); - await service.request('owner', 'owner-revision', query, true); - await settled(); - for (let i = 0; i < 5; i++) await service.read('owner', query, true); - expect(await repository.read(query)).toEqual(before); - expect((await service.read('owner', query, true)).meta.sync.state).toBe( - 'failed', - ); - expect(upstream.findShares).toHaveBeenCalledTimes(2); - expect((await service.read('other', query)).meta.sync.errorCode).toBeNull(); - }); - it('isolates invalid credentials from shared empty lookup and another user', async () => { - const blocked = deferred(); - upstream.findShares.mockReturnValueOnce(blocked.promise); - await service.read('owner', query); - await service.read('other', query); - blocked.reject(new Error('INVALID_TBANK_TOKEN')); - await settled(); - expect(await prisma.tbankCatalogDiscovery.count()).toBe(0); - expect( - ( - await prisma.tbankConnection.findUniqueOrThrow({ - where: { userId: 'owner' }, - }) - ).invalidToken, - ).toBe(true); - expect((await service.read('other', query, true)).meta.sync.errorCode).toBe( - 'TBANK_UNAVAILABLE', - ); - await service.read('other', query); - await settled(); - expect((await service.read('other', query, true)).data).toEqual([ - instrument, - ]); - }); - it('recovers abandoned work only on the next ordinary request and rejects superseded attempts', async () => { - const abandoned = await repository.reserve( - { userId: 'owner', revision: 'owner-revision' }, - query, - 'old-process', - new Set(), - ); - assert(abandoned); - service = createService(); - await service.read('owner', query, true); - await delay(10); expect(upstream.findShares).not.toHaveBeenCalled(); - await service.read('owner', query); - await settled(); - expect( - await repository.publish( - { - userId: 'owner', - revision: 'owner-revision', - id: abandoned.work.id, - attemptId: abandoned.work.attemptId, - }, - query, - [], - new Date(), - ), - ).toBe(false); - expect((await service.read('owner', query, true)).data).toEqual([ - instrument, - ]); }); - it('keeps known discovery stable between refresh requests', async () => { - await service.read('owner', query); - await settled(); - const now = Date.now(); - const clock = jest.spyOn(Date, 'now').mockReturnValue(now); - try { - await prisma.tbankCatalogDiscovery.updateMany({ - data: { fetchedAt: new Date(now - 86_399_000) }, - }); - await service.read('other', query); - expect(upstream.findShares).toHaveBeenCalledTimes(1); - await prisma.tbankCatalogDiscovery.updateMany({ - data: { fetchedAt: new Date(now - 86_400_000) }, - }); - expect((await service.read('other', query, true)).meta.freshness).toBe('fresh'); - expect(upstream.findShares).toHaveBeenCalledTimes(1); - await service.read('other', query); - expect(upstream.findShares).toHaveBeenCalledTimes(1); - } finally { - clock.mockRestore(); - } + it('keeps a confirmed empty result distinct from an unknown query', async () => { + await seedCatalog('GAZP', []); + const empty = await service.read('owner', { ...query, query: 'GAZP' }); + expect(empty.data).toEqual([]); + expect(empty.meta.availability).toBe('available'); + const unknown = await service.read('owner', { ...query, query: 'LKOH' }); + expect(unknown.data).toBeNull(); + }); + it('keeps the stored selection order and does not expose private fields', async () => { + const first = { ...instrument, uid: 'first', name: 'First' }; + const second = { ...instrument, uid: 'second', name: 'Second' }; + await seedCatalog('SBER', [ + { uid: 'second', payload: second }, + { uid: 'first', payload: first }, + ]); + const result = await service.read('owner', query); + assert(result.data); + expect(result.data.map((item) => item.uid)).toEqual(['second', 'first']); + expect(result.data[0]).not.toHaveProperty('accountId'); }); }); -function deferred() { - let resolve: (value: T) => void = () => { - throw new Error('Uninitialized deferred'); - }; - let reject: (error: Error) => void = () => { - throw new Error('Uninitialized deferred'); - }; - const promise = new Promise((yes, no) => { - resolve = yes; - reject = no; - }); - return { promise, resolve, reject }; -} -- 2.54.0