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'; import { CatalogDiscoveryService } from '../src/modules/investments/instruments/application/catalog-discovery.service'; import { CatalogDiscoveryRepository, type CatalogInstrument, type CatalogQuery, } 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. describe('shared catalog discovery (SQLite)', () => { const prisma = new PrismaService( new ConfigService({ DATABASE_URL: process.env.DATABASE_URL }), ); const tokens = new TbankTokenService( new ConfigService({ TOKEN_ENCRYPTION_KEY: process.env.TOKEN_ENCRYPTION_KEY, }) as never, ); const upstream = { findShares: jest.fn(), findBonds: jest.fn() }; const access = new TbankAccessService(prisma, tokens); const repository = new CatalogDiscoveryRepository(prisma); const createService = () => new CatalogDiscoveryService( repository, access, upstream as never, { event: jest.fn() } as never, { requestMany: jest.fn().mockResolvedValue([]) } as never, ); let service: CatalogDiscoveryService; const query: CatalogQuery = { instrumentType: 'share', query: 'SBER' }; const instrument = { uid: 'uid', name: 'Synthetic share', ticker: 'SBER', isin: 'TEST', instrumentKind: 'INSTRUMENT_TYPE_SHARE', }; const seed = (userId: string) => prisma.tbankConnection.create({ data: { userId, revision: `${userId}-revision`, encryptedToken: tokens.encrypt(`synthetic-${userId}`), 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); } throw new Error('Catalog work did not settle'); } beforeAll(async () => { await prisma.$connect(); for (const id of ['owner', 'other']) { await prisma.user.create({ data: { id, email: `${id}@example.invalid`, passwordHash: 'synthetic' }, }); } }); afterAll(async () => prisma.$disconnect()); beforeEach(async () => { await prisma.tbankConnection.deleteMany(); await prisma.tbankCatalogDiscovery.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 () => { const cold = await service.read('owner', { ...query, query: ' sber ' }); expect(cold.data).toBeNull(); expect(cold.meta.availability).toBe('missing'); await settled(); await prisma.$disconnect(); service = createService(); const decrypt = jest.spyOn(tokens, 'decrypt'); const warm = await service.read('other', query); expect(warm.data).toEqual([instrument]); expect(warm.meta.availability).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(); } }); }); 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 }; }