336 lines
12 KiB
TypeScript
336 lines
12 KiB
TypeScript
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<CatalogInstrument[]>();
|
|
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<CatalogInstrument[]>();
|
|
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<CatalogInstrument[]>();
|
|
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<T>() {
|
|
let resolve: (value: T) => void = () => {
|
|
throw new Error('Uninitialized deferred');
|
|
};
|
|
let reject: (error: Error) => void = () => {
|
|
throw new Error('Uninitialized deferred');
|
|
};
|
|
const promise = new Promise<T>((yes, no) => {
|
|
resolve = yes;
|
|
reject = no;
|
|
});
|
|
return { promise, resolve, reject };
|
|
}
|