import { ConfigService } from '@nestjs/config'; 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 { CandleService } from '../src/modules/investments/instruments/application/candle.service'; import { candleGaps, type CandleRange, } from '../src/modules/investments/instruments/domain/candle-coverage'; import { CandleRepository } from '../src/modules/investments/instruments/infrastructure/candle.repository'; import { PrismaService } from '../src/prisma/prisma.service'; describe('atomic shared candle reconciliation (SQLite)', () => { const prisma = new PrismaService( new ConfigService({ DATABASE_URL: process.env.DATABASE_URL }), ); const repository = new CandleRepository(prisma); const owner = { userId: 'owner', revision: 'revision' }; const tokens = new TbankTokenService( new ConfigService({ TOKEN_ENCRYPTION_KEY: process.env.TOKEN_ENCRYPTION_KEY, }) as never, ); const upstream = { getCandles: jest.fn() }; const access = new TbankAccessService(prisma, tokens); const instrument = { uid: 'uid', currency: 'rub', first_1dayCandleDate: { seconds: String(Date.parse('2020-01-01Z') / 1000) }, }; const sources = { ensureDetail: jest.fn().mockResolvedValue(instrument), readMany: jest.fn().mockResolvedValue([ { input: { uid: 'uid', instrumentType: 'share', kind: 'detail' }, instrument, meta: { availability: 'complete', freshness: 'fresh', lastSuccessfulAt: null, resourceKey: 'detail/uid', dataRevision: '1', sync: { id: null, state: 'idle', errorCode: null }, }, }, ]), }; const createService = (selectedSources = sources) => new CandleService( repository, selectedSources as never, access, upstream as never, { event: jest.fn() } as never, ); async function settled() { for (let i = 0; i < 500; i++) { if (!(await prisma.tbankSyncWork.count({ where: { state: 'running' } }))) { return; } await delay(5); } throw new Error('Candle work did not settle'); } const range = { from: new Date('2026-01-01Z'), to: new Date('2026-01-04Z') }; const candle = { time: '2026-01-02T00:00:00.000Z', close: '100.000000001', isComplete: true, }; async function reserve(ranges: CandleRange[] = [range]) { const result = await repository.reserve( owner, 'candles:uid', { uid: 'uid', ranges }, 'run', new Set(), ); if (!result) throw new Error('Missing attempt'); return { ...owner, id: result.work.id, attemptId: result.work.attemptId }; } beforeAll(async () => { await prisma.user.create({ data: { id: 'owner', email: 'owner@example.invalid', passwordHash: 'synthetic', }, }); }); beforeEach(async () => { jest.clearAllMocks(); upstream.getCandles.mockReset().mockResolvedValue([]); await prisma.tbankConnection.deleteMany(); await prisma.tbankInstrument.deleteMany(); await prisma.tbankConnection.create({ data: { ...owner, encryptedToken: tokens.encrypt('synthetic'), connectedAt: new Date(), }, }); await prisma.tbankInstrument.create({ data: { provider: 'tbank', uid: 'uid', instrumentType: 'share', metadata: {}, fetchedAt: new Date(), }, }); }); afterAll(async () => prisma.$disconnect()); afterEach(() => jest.restoreAllMocks()); it('deduplicates seams, excludes outside rows, persists exact decimals and closed empty coverage after reopen', async () => { const attempt = await reserve(); await repository.stage( attempt, range, [candle, candle, { ...candle, time: range.to.toISOString() }], new Date('2026-01-05Z'), ); expect((await repository.read('uid', range)).data).toEqual([]); await repository.publish(attempt, { uid: 'uid', ranges: [range] }); await prisma.$disconnect(); const read = await new CandleRepository(prisma).read('uid', range); expect(read.data).toEqual([candle]); expect(candleGaps(range, read.coverage, new Date('2030-01-01Z'))).toEqual( [], ); const emptyRange = { from: range.to, to: new Date('2026-01-06Z') }; const empty = await reserve([emptyRange]); await repository.stage(empty, emptyRange, [], new Date('2026-01-07Z')); await repository.publish(empty, { uid: 'uid', ranges: [emptyRange] }); const readEmpty = await repository.read('uid', emptyRange); expect(readEmpty.data).toEqual([]); expect( candleGaps(emptyRange, readEmpty.coverage, new Date('2030-01-01Z')), ).toEqual([]); }); it('keeps the entire previous snapshot and timestamps when a later refresh window fails', async () => { const initial = await reserve(); await repository.stage(initial, range, [candle], new Date('2026-01-05Z')); await repository.publish(initial, { uid: 'uid', ranges: [range] }); const before = await repository.read('uid', range); const windows = [ { from: range.from, to: new Date('2026-01-03Z') }, { from: new Date('2026-01-03Z'), to: range.to }, ]; const retry = await reserve(windows); await repository.stage( retry, windows[0], [{ ...candle, close: '2' }], new Date('2026-01-06Z'), ); await expect( repository.publish(retry, { uid: 'uid', ranges: windows }), ).rejects.toThrow('Incomplete candle reconciliation'); expect(await repository.read('uid', range)).toEqual(before); await repository.fail(retry, 'TBANK_UNAVAILABLE'); expect(await repository.read('uid', range)).toEqual(before); expect(await prisma.tbankSyncStage.count()).toBe(0); }); it('reconciles deleted rows atomically and preserves coverage outside the requested period', async () => { const initial = await reserve(); await repository.stage(initial, range, [candle], new Date('2026-01-05Z')); await repository.publish(initial, { uid: 'uid', ranges: [range] }); const middle = { from: new Date('2026-01-02Z'), to: new Date('2026-01-03Z'), }; const retry = await reserve([middle]); await repository.stage(retry, middle, [], new Date('2026-01-06Z')); expect((await repository.read('uid', range)).data).toEqual([candle]); await repository.publish(retry, { uid: 'uid', ranges: [middle] }); const after = await repository.read('uid', range); expect(after.data).toEqual([]); expect(candleGaps(range, after.coverage, new Date('2030-01-01Z'))).toEqual( [], ); }); it('fences superseded and deleted attempts including shared coverage timestamps', async () => { const old = await reserve(); await repository.stage(old, range, [candle], new Date('2026-01-05Z')); const current = await reserve(); expect(await repository.publish(old, { uid: 'uid', ranges: [range] })).toBe( false, ); expect( await repository.stage(old, range, [candle], new Date('2026-01-06Z')), ).toBe(false); await repository.stage(current, range, [candle], new Date('2026-01-06Z')); await prisma.tbankConnection.deleteMany(); expect( await repository.publish(current, { uid: 'uid', ranges: [range] }), ).toBe(false); expect((await repository.read('uid', range)).coverage).toEqual([]); }); it('coalesces twenty reads and reuses closed windows after reopen', async () => { const service = createService(); const input = { uid: 'uid', instrumentType: 'share' as const, period: 'all' as const, }; let release: (value: unknown[]) => void = () => {}; const blocked = new Promise((resolve) => { release = resolve; }); upstream.getCandles.mockReturnValueOnce(blocked); const cold = await Promise.all( Array.from({ length: 20 }, () => service.read('owner', input)), ); expect(cold.every((row) => row.data === null)).toBe(true); expect(await prisma.tbankSyncWork.count()).toBe(1); release([]); await settled(); const count = upstream.getCandles.mock.calls.length; expect(count).toBeGreaterThan(1); for (const call of upstream.getCandles.mock.calls as [ string, string, Date, Date, ][]) { expect(call[3].valueOf() - call[2].valueOf()).toBeLessThanOrEqual( 365 * 86400000, ); } await prisma.$disconnect(); const reopened = createService(); expect((await reopened.read('owner', input)).data?.candles).toEqual([]); expect(upstream.getCandles).toHaveBeenCalledTimes(count); await prisma.tbankCandleCoverage.updateMany({ where: { isComplete: false }, data: { fetchedAt: new Date(0) }, }); const stale = await reopened.read('owner', input); expect(stale.meta.availability).toBe('available'); expect(stale.meta.completeness).toBe('complete'); expect(stale.meta.freshness).toBe('stale'); expect(stale.meta.coverage?.missingRanges).toEqual([]); expect(stale.meta.coverage?.completeRanges.length).toBeGreaterThan(0); expect(stale.meta.coverage?.historyState).toBe('complete'); expect(upstream.getCandles).toHaveBeenCalledTimes(count); await reopened.read('owner', input); await settled(); expect(upstream.getCandles).toHaveBeenCalledTimes(count + 1); }); it('recovers abandoned work on an ordinary request', async () => { const input = { uid: 'uid', instrumentType: 'share' as const, period: 'month' as const, }; const abandoned = await repository.reserve( owner, 'candles:uid:month', { uid: 'uid', ranges: [] }, 'old-process', new Set(), ); const reopened = createService(); await reopened.read('owner', input); await settled(); expect(upstream.getCandles).toHaveBeenCalledTimes(1); expect( (await repository.work('owner', 'revision', 'candles:uid:month')) ?.attemptId, ).not.toBe(abandoned?.work.attemptId); }); it('retains full shown history when a forced reread fails in its second window', async () => { const input = { uid: 'uid', instrumentType: 'share' as const, period: 'all' as const, }; const service = createService(); await service.read('owner', input); await settled(); const before = await prisma.tbankCandleCoverage.findMany(); upstream.getCandles .mockResolvedValueOnce([ { time: { seconds: '1577923200' }, close: { units: '123', nano: 1 }, isComplete: true, }, ]) .mockRejectedValueOnce(new Error('second window unavailable')); await service.request('owner', 'revision', input, new Date(), true); await settled(); expect(await prisma.tbankCandleCoverage.findMany()).toEqual(before); expect(await prisma.tbankCandle.count()).toBe(0); const read = await service.read('owner', input); expect(read.meta.lastAttempt.state).toBe('failed'); expect(read.data?.candles).toEqual([]); }); it.each([ ['owner', 'before loading', false], ['observer', 'before loading', false], ['owner', 'after ordinary publication', false], ['observer', 'after ordinary publication', false], ['observer', 'after ordinary publication', true], ] as const)( 'honors force from %s joining %s (later failure=%s)', async (userId, phase, laterFailure) => { if (userId !== owner.userId) { await prisma.user.upsert({ where: { id: userId }, create: { id: userId, email: `${userId}@example.invalid`, passwordHash: 'synthetic', }, update: {}, }); await prisma.tbankConnection.create({ data: { userId, revision: 'revision', encryptedToken: tokens.encrypt('synthetic'), connectedAt: new Date(), }, }); } const history = { ...instrument, first_1dayCandleDate: { seconds: String(Date.parse('2024-01-01Z') / 1000), }, }; const covered = { from: new Date('2024-01-01Z'), to: new Date('2025-01-01Z'), }; const old = { ...candle, time: '2024-01-02T00:00:00.000Z' }; const initial = await reserve([covered]); await repository.stage(initial, covered, [old], new Date('2025-01-02Z')); await repository.publish(initial, { uid: 'uid', ranges: [covered] }); let entered = () => {}; let release = () => {}; const started = new Promise((resolve) => (entered = resolve)); const blocked = new Promise((resolve) => (release = resolve)); const selectedSources = { ...sources, ensureDetail: jest.fn().mockResolvedValue(history), }; if (phase === 'before loading') { selectedSources.ensureDetail.mockImplementationOnce(async () => { entered(); await blocked; return history; }); } else { const publish = repository.publish.bind(repository); jest .spyOn(repository, 'publish') .mockImplementationOnce(async (...args) => { const result = await publish(...args); entered(); await blocked; return result; }); } const forcedAt = new Date('2026-02-03T13:00:00Z'); const forcedStates: string[] = []; const fullRange = { from: covered.from, to: new Date('2026-02-04Z') }; let beforeForced: Awaited> | undefined; upstream.getCandles.mockImplementation( async (_token, _uid, _from, to: Date) => { if (laterFailure && forcedStates.length) { throw new Error('Forced reconciliation second window unavailable'); } if (to.valueOf() === forcedAt.valueOf()) { beforeForced = await repository.read('uid', fullRange); const work = await repository.work( userId, 'revision', 'candles:uid:all', ); forcedStates.push(work?.state ?? 'missing'); } return [ { time: { seconds: String(Date.parse(old.time) / 1000) }, close: { units: '200', nano: 1 }, isComplete: true, }, ]; }, ); const service = createService(selectedSources); const input = { uid: 'uid', instrumentType: 'share' as const, period: 'all' as const, }; const ordinaryId = await service.request( 'owner', 'revision', input, new Date('2026-02-02T12:00:00Z'), false, 2, ); await started; const forcedId = await service.request( userId, 'revision', input, forcedAt, true, 0, ); release(); await settled(); if (laterFailure) { // eslint-disable-next-line jest/no-conditional-expect -- The parameterized fixture chooses the scenario; each branch asserts its outcome and the test also asserts common invariants. expect(beforeForced).toBeDefined(); // eslint-disable-next-line jest/no-conditional-expect -- The parameterized fixture chooses the scenario; each branch asserts its outcome and the test also asserts common invariants. expect(await repository.read('uid', fullRange)).toEqual(beforeForced); } else { // eslint-disable-next-line jest/no-conditional-expect -- The parameterized fixture chooses the scenario; each branch asserts its outcome and the test also asserts common invariants. expect((await repository.read('uid', covered)).data).toEqual([ { ...old, close: '200.000000001' }, ]); } expect(forcedStates).toEqual(['running']); expect( await repository.work(userId, 'revision', 'candles:uid:all'), ).toMatchObject({ id: forcedId, state: laterFailure ? 'failed' : 'complete', priority: 0, intent: { force: true, now: forcedAt.toISOString(), }, }); // eslint-disable-next-line jest/no-conditional-expect -- The parameterized fixture chooses the scenario; each branch asserts its outcome and the test also asserts common invariants. if (userId === owner.userId) expect(forcedId).toBe(ordinaryId); // eslint-disable-next-line jest/no-conditional-expect -- The parameterized fixture chooses the scenario; each branch asserts its outcome and the test also asserts common invariants. else expect(forcedId).not.toBe(ordinaryId); }, ); it('preserves source incompleteness and bond percentage precision without nominal conversion', async () => { const now = new Date(); now.setUTCHours(0, 0, 0, 0); upstream.getCandles.mockResolvedValue([ { time: { seconds: String(now.valueOf() / 1000) }, close: { units: '99', nano: 123456789 }, isComplete: false, }, ]); const service = createService(); const input = { uid: 'uid', instrumentType: 'bond' as const, period: 'month' as const, }; await service.read('owner', input); await settled(); const result = await service.read('owner', input); expect(result.data?.currency).toBeNull(); expect(result.data?.candles).toEqual([ { time: now.toISOString(), close: '99.123456789', isComplete: false }, ]); expect( await prisma.tbankCandleCoverage.count({ where: { isComplete: false } }), ).toBe(1); }); it('loads only the requested period and reuses its saved coverage', async () => { const service = createService(); const input = { uid: 'uid', instrumentType: 'share', period: 'month', } as const; await service.read('owner', input); await settled(); expect(upstream.getCandles).toHaveBeenCalledTimes(1); expect(await repository.work('owner', 'revision', 'candles:uid:all')).toBeNull(); const count = upstream.getCandles.mock.calls.length; await service.read('owner', input); await settled(); expect(upstream.getCandles).toHaveBeenCalledTimes(count); }); });