503 lines
18 KiB
TypeScript
503 lines
18 KiB
TypeScript
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, makes observe passive 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<unknown[]>((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);
|
|
await service.read('owner', input, true);
|
|
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, true);
|
|
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 only on a new 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, true);
|
|
expect(upstream.getCandles).not.toHaveBeenCalled();
|
|
expect(
|
|
(await repository.work('owner', 'revision', 'candles:uid:month'))
|
|
?.attemptId,
|
|
).toBe(abandoned?.work.attemptId);
|
|
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, true);
|
|
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<void>((resolve) => (entered = resolve));
|
|
const blocked = new Promise<void>((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<ReturnType<CandleRepository['read']>> | 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, true);
|
|
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);
|
|
});
|
|
});
|