import assert from 'node:assert/strict'; import { ConfigService } from '@nestjs/config'; import { setTimeout as delay } from 'node:timers/promises'; import { PrismaService } from '../src/prisma/prisma.service'; import { TbankAccessService } from '../src/integrations/tbank/tbank-access.service'; import { TbankTokenService } from '../src/integrations/tbank/tbank-token.service'; import { InstrumentMetadataService } from '../src/integrations/tbank/instrument-metadata.service'; import { InstrumentMetadataRepository, metadataKey, } from '../src/integrations/tbank/instrument-metadata.repository'; import type { TbankInstrument } from '../src/integrations/tbank/tbank.client'; // Shared publication is proven against real FK/transaction/connection boundaries. describe('durable shared instrument metadata (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 = { getInstrumentByUid: jest.fn(), getInstrumentByFigi: jest.fn(), }; const access = new TbankAccessService(prisma, tokens, upstream as never); const repository = new InstrumentMetadataRepository(prisma); const logger = { event: jest.fn() }; const createService = () => new InstrumentMetadataService( upstream as never, logger as never, repository, access, ); let service: InstrumentMetadataService; const identity = { instrumentUid: 'uid', figi: 'FIGI' }; const known = { uid: 'uid', figi: 'FIGI', ticker: 'TEST', name: 'Synthetic instrument', brand: { logoName: 'test.png' }, }; const seedConnection = (userId: string) => prisma.tbankConnection.create({ data: { userId, revision: `${userId}-revision`, encryptedToken: tokens.encrypt(`synthetic-${userId}`), connectedAt: new Date(), }, }); async function settled() { for (let index = 0; index < 300; index++) { if (!(await prisma.tbankSyncWork.count({ where: { state: 'running' } }))) return; await delay(5); } throw new Error('Metadata 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.tbankInstrumentLookup.deleteMany(); await prisma.tbankInstrument.deleteMany(); await seedConnection('owner'); await seedConnection('other'); jest.resetAllMocks(); upstream.getInstrumentByUid.mockResolvedValue(known); upstream.getInstrumentByFigi.mockResolvedValue(known); service = createService(); }); it('reuses durable successful metadata across users and reopen without decrypt or RPC', async () => { const cold = await service.read('owner', identity); expect(cold?.data).toBeNull(); expect(cold?.work?.state).toBe('running'); await settled(); await prisma.$disconnect(); service = createService(); const decrypt = jest.spyOn(tokens, 'decrypt'); const other = await service.read('other', identity); expect(other?.data).toEqual(known); expect(other?.fresh).toBe(true); expect(other?.work).toBeNull(); expect((await service.read('other', { figi: 'FIGI' }))?.data).toEqual( known, ); expect(upstream.getInstrumentByUid).toHaveBeenCalledTimes(1); expect(upstream.getInstrumentByFigi).not.toHaveBeenCalled(); expect(decrypt).not.toHaveBeenCalled(); decrypt.mockRestore(); const rows = await prisma.tbankInstrument.findMany(); expect(JSON.stringify(rows)).not.toMatch( /synthetic-owner|owner-revision|quantity|estimatedAmount/, ); }); it('persists UID miss separately from successful FIGI fallback and confirmed empty lookups', async () => { upstream.getInstrumentByUid.mockResolvedValue(null); await service.read('owner', identity); await settled(); const result = await service.read('other', identity); expect(result?.data).toEqual(known); expect(result?.complete).toBe(true); expect(upstream.getInstrumentByUid).toHaveBeenCalledTimes(1); expect(upstream.getInstrumentByFigi).toHaveBeenCalledTimes(1); upstream.getInstrumentByFigi.mockResolvedValue(null); const missing = { figi: 'UNKNOWN' }; await service.read('owner', missing); await settled(); const empty = await service.read('other', missing); expect(empty?.data).toBeNull(); expect(empty?.complete).toBe(true); expect(empty?.fresh).toBe(true); expect(upstream.getInstrumentByFigi).toHaveBeenCalledTimes(2); }); it('coalesces twenty requests across users while giving each user a private handle', async () => { const blocked = deferred(); upstream.getInstrumentByUid.mockReturnValue(blocked.promise); const responses = await Promise.all( Array.from({ length: 20 }, (_, i) => service.read(i % 2 ? 'owner' : 'other', identity), ), ); await delay(10); expect(upstream.getInstrumentByUid).toHaveBeenCalledTimes(1); const ids = new Set(responses.map((response) => response?.work?.id)); expect(ids.size).toBe(2); expect(responses[0]?.work?.id).not.toBe(responses[1]?.work?.id); blocked.resolve(known); await settled(); expect(await prisma.tbankInstrument.count()).toBe(1); expect( await prisma.tbankSyncWork.count({ where: { state: 'complete' } }), ).toBe(2); }); it.each(['delete', 'replace'])( 'discards a producer after %s, including shared freshness, without borrowing another user token', async (action) => { const blocked = deferred(); upstream.getInstrumentByUid.mockReturnValueOnce(blocked.promise); await service.read('owner', identity); await delay(10); await service.read('other', identity); await prisma.tbankConnection.delete({ where: { userId: 'owner' } }); if (action === 'replace') await prisma.tbankConnection.create({ data: { userId: 'owner', revision: 'replacement', encryptedToken: tokens.encrypt('replacement-token'), connectedAt: new Date(), }, }); blocked.resolve(known); await settled(); expect(await prisma.tbankInstrument.count()).toBe(0); expect(await prisma.tbankInstrumentLookup.count()).toBe(0); expect( (await service.read('other', identity, true))?.work?.errorCode, ).toBe('TBANK_UNAVAILABLE'); expect(upstream.getInstrumentByUid).toHaveBeenCalledTimes(1); await service.read('other', identity); await settled(); expect((await service.read('other', identity, true))?.data).toEqual( known, ); expect(upstream.getInstrumentByUid).toHaveBeenLastCalledWith( 'synthetic-other', 'uid', ); }, ); it('does not turn initiating credentials failure into a shared negative lookup or a foreign credentials error', async () => { const blocked = deferred(); upstream.getInstrumentByUid.mockReturnValueOnce(blocked.promise); await service.read('owner', identity); await delay(10); await service.read('other', identity); blocked.reject(new Error('INVALID_TBANK_TOKEN')); await settled(); await expect(service.read('owner', identity, true)).rejects.toMatchObject({ response: { code: 'INVALID_TBANK_TOKEN' }, }); expect((await service.read('other', identity, true))?.work?.errorCode).toBe( 'TBANK_UNAVAILABLE', ); expect(await prisma.tbankInstrumentLookup.count()).toBe(0); await service.read('other', identity); await settled(); expect((await service.read('other', identity, true))?.data).toEqual(known); }); it('keeps last good metadata and source time after refresh failure; stale observation does not restart work', async () => { await service.read('owner', identity); await settled(); const old = new Date(Date.now() - 86_400_000); await prisma.tbankInstrumentLookup.updateMany({ data: { fetchedAt: old } }); await prisma.tbankInstrument.updateMany({ data: { fetchedAt: old } }); const calls = upstream.getInstrumentByUid.mock.calls.length; expect((await service.read('owner', identity, true))?.fresh).toBe(false); expect(upstream.getInstrumentByUid).toHaveBeenCalledTimes(calls); upstream.getInstrumentByUid.mockRejectedValueOnce( new Error('upstream unavailable'), ); upstream.getInstrumentByFigi.mockRejectedValueOnce( new Error('upstream unavailable'), ); await service.read('owner', identity); await settled(); const failed = await service.read('owner', identity, true); expect(failed?.data).toEqual(known); expect(failed?.fetchedAt).toEqual(old); expect(failed?.work?.state).toBe('failed'); await delay(20); expect(upstream.getInstrumentByUid).toHaveBeenCalledTimes(calls + 1); }); it('coalesces forced refresh and retains freshness/timestamp when it fails before the TTL', async () => { await service.read('owner', identity); await settled(); const before = await service.read('owner', identity, true); const blocked = deferred(); upstream.getInstrumentByUid.mockReturnValueOnce(blocked.promise); upstream.getInstrumentByFigi.mockRejectedValueOnce( new Error('also unavailable'), ); const handles = await Promise.all( Array.from({ length: 20 }, () => service.request('owner', 'owner-revision', identity, true), ), ); expect(new Set(handles).size).toBe(1); blocked.reject(new Error('unavailable')); await settled(); const after = await service.read('owner', identity, true); expect(after?.data).toEqual(before?.data); expect(after?.fetchedAt).toEqual(before?.fetchedAt); expect(after?.fresh).toBe(true); expect(after?.work?.state).toBe('failed'); expect(upstream.getInstrumentByUid).toHaveBeenCalledTimes(2); }); it('uses the exact daily TTL boundary and does not count another instrument as covered', async () => { await service.read('owner', identity); await settled(); const timestamp = new Date('2026-09-01Z'); await prisma.tbankInstrumentLookup.updateMany({ data: { fetchedAt: timestamp }, }); const clock = jest .spyOn(Date, 'now') .mockReturnValue(timestamp.valueOf() + 86_400_000 - 1); try { expect((await service.read('owner', identity, true))?.fresh).toBe(true); clock.mockReturnValue(timestamp.valueOf() + 86_400_000); expect((await service.read('owner', identity, true))?.fresh).toBe(false); expect( (await service.read('owner', { instrumentUid: 'other-uid' }, true)) ?.complete, ).toBe(false); expect(upstream.getInstrumentByUid).toHaveBeenCalledTimes(1); } finally { clock.mockRestore(); } }); it('recovers abandoned work only on a new request after reopen, with a new attempt', async () => { const old = await repository.reserve( { userId: 'owner', revision: 'owner-revision' }, identity, 'previous-process', new Set(), ); assert(old); await prisma.$disconnect(); service = createService(); await service.read('owner', identity, true); await delay(10); expect(upstream.getInstrumentByUid).not.toHaveBeenCalled(); await service.read('owner', identity); await settled(); const work = await prisma.tbankSyncWork.findUniqueOrThrow({ where: { id: old.work.id }, }); expect(work.attemptId).not.toBe(old.work.attemptId); expect(work.state).toBe('complete'); expect(upstream.getInstrumentByUid).toHaveBeenCalledTimes(1); }); it('rejects publication by a superseded attempt without changing payload or successful timestamp', async () => { const owner = { userId: 'owner', revision: 'owner-revision' }; const old = await repository.reserve(owner, identity, 'old', new Set()); assert(old); const current = await repository.reserve( owner, identity, 'current', new Set(), ); assert(current); const fence = (work: typeof old.work) => ({ ...owner, id: work.id, attemptId: work.attemptId, }); expect( await repository.publish( fence(current.work), [{ type: 'uid', id: 'uid', data: known }], new Date(1000), ), ).toBe(true); expect( await repository.publish( fence(old.work), [{ type: 'uid', id: 'uid', data: { ...known, name: 'late' } }], new Date(2000), ), ).toBe(false); const saved = (await repository.readMany([identity])).get( metadataKey(identity), ); expect(saved?.data).toEqual(known); expect(saved?.fetchedAt).toEqual(new Date(1000)); }); it('uses a verified FIGI fallback after transient UID failure without persisting a false UID miss', async () => { upstream.getInstrumentByUid.mockRejectedValueOnce( new Error('temporarily unavailable'), ); await service.read('owner', identity); await settled(); const result = await service.read('other', identity, true); expect(result?.data).toEqual(known); expect(result?.complete).toBe(true); const lookup = await prisma.tbankInstrumentLookup.findFirstOrThrow({ where: { identifierType: 'uid', identifier: 'uid' }, }); expect(lookup.matchedUids).toEqual(['uid']); }); it('reads a full 1000-row ledger metadata batch without SQLite expression-depth overflow', async () => { await service.read('owner', identity); await settled(); const identities = Array.from({ length: 1000 }, (_, index) => index ? { instrumentUid: `uid-${index}`, figi: `FIGI-${index}` } : identity, ); const saved = await service.savedMany(identities); expect(saved.size).toBe(1000); expect(saved.get(metadataKey(identity))?.data).toEqual(known); expect(upstream.getInstrumentByUid).toHaveBeenCalledTimes(1); }); it('does not cache a sparse malformed UID result as complete metadata', async () => { upstream.getInstrumentByUid.mockResolvedValue({ uid: 'uid', name: ' ' }); await service.read('owner', { instrumentUid: 'uid' }); await settled(); const result = await service.read('owner', { instrumentUid: 'uid' }, true); expect(result?.complete).toBe(false); expect(result?.work?.state).toBe('failed'); expect(await prisma.tbankInstrumentLookup.count()).toBe(0); }); it('does not fabricate a provider UID from a malformed FIGI response', async () => { upstream.getInstrumentByFigi.mockResolvedValue({ name: 'No provider identity', }); await service.read('owner', { figi: 'FIGI' }); await settled(); expect(await prisma.tbankInstrument.count()).toBe(0); expect(await prisma.tbankInstrumentLookup.count()).toBe(0); expect( (await service.read('owner', { figi: 'FIGI' }, true))?.work?.state, ).toBe('failed'); }); }); 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 }; }