Merge branch 'develop' into test/65-isolate-http-suites
This commit is contained in:
@@ -1,6 +1,5 @@
|
||||
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';
|
||||
@@ -12,7 +11,7 @@ import {
|
||||
} 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.
|
||||
// Real transactions prove query normalization, confirmed-empty identity and cross-user isolation of shared reads.
|
||||
describe('shared catalog discovery (SQLite)', () => {
|
||||
const prisma = new PrismaService(
|
||||
new ConfigService({ DATABASE_URL: process.env.DATABASE_URL }),
|
||||
@@ -30,7 +29,6 @@ describe('shared catalog discovery (SQLite)', () => {
|
||||
repository,
|
||||
access,
|
||||
upstream as never,
|
||||
{ event: jest.fn() } as never,
|
||||
{ requestMany: jest.fn().mockResolvedValue([]) } as never,
|
||||
);
|
||||
let service: CatalogDiscoveryService;
|
||||
@@ -51,14 +49,42 @@ describe('shared catalog discovery (SQLite)', () => {
|
||||
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);
|
||||
// Writes a finished discovery directly, so the read path is tested without any producer.
|
||||
async function seedCatalog(
|
||||
query: string,
|
||||
instruments: Array<{ payload: CatalogInstrument; uid: string }>,
|
||||
) {
|
||||
for (const item of instruments) {
|
||||
await prisma.tbankInstrument.create({
|
||||
data: {
|
||||
provider: 'tbank',
|
||||
uid: item.uid,
|
||||
instrumentType: 'share',
|
||||
metadata: {},
|
||||
fetchedAt: new Date(),
|
||||
},
|
||||
});
|
||||
await prisma.tbankInstrumentSource.create({
|
||||
data: {
|
||||
provider: 'tbank',
|
||||
instrumentUid: item.uid,
|
||||
source: 'tbank',
|
||||
kind: 'catalog',
|
||||
rangeKey: 'share',
|
||||
payload: item.payload,
|
||||
fetchedAt: new Date(),
|
||||
},
|
||||
});
|
||||
}
|
||||
throw new Error('Catalog work did not settle');
|
||||
await prisma.tbankCatalogDiscovery.create({
|
||||
data: {
|
||||
provider: 'tbank',
|
||||
instrumentType: 'share',
|
||||
query,
|
||||
instrumentUids: instruments.map((item) => item.uid),
|
||||
fetchedAt: new Date(),
|
||||
},
|
||||
});
|
||||
}
|
||||
beforeAll(async () => {
|
||||
await prisma.$connect();
|
||||
@@ -72,265 +98,41 @@ describe('shared catalog discovery (SQLite)', () => {
|
||||
beforeEach(async () => {
|
||||
await prisma.tbankConnection.deleteMany();
|
||||
await prisma.tbankCatalogDiscovery.deleteMany();
|
||||
await prisma.tbankInstrumentSource.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 () => {
|
||||
it('treats an unknown query as unavailable and a seeded query as available for another user', async () => {
|
||||
const cold = await service.read('owner', { ...query, query: ' sber ' });
|
||||
expect(cold.data).toBeNull();
|
||||
expect(cold.meta.availability).toBe('initial-loading');
|
||||
await settled();
|
||||
await prisma.$disconnect();
|
||||
service = createService();
|
||||
const decrypt = jest.spyOn(tokens, 'decrypt');
|
||||
expect(cold.meta.availability).toBe('temporarily-unavailable');
|
||||
await seedCatalog('SBER', [{ uid: 'uid', payload: instrument }]);
|
||||
const warm = await service.read('other', query);
|
||||
expect(warm.data).toEqual([instrument]);
|
||||
expect(warm.meta.availability).toBe('available');
|
||||
expect(warm.meta.completeness).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();
|
||||
}
|
||||
it('keeps a confirmed empty result distinct from an unknown query', async () => {
|
||||
await seedCatalog('GAZP', []);
|
||||
const empty = await service.read('owner', { ...query, query: 'GAZP' });
|
||||
expect(empty.data).toEqual([]);
|
||||
expect(empty.meta.availability).toBe('available');
|
||||
const unknown = await service.read('owner', { ...query, query: 'LKOH' });
|
||||
expect(unknown.data).toBeNull();
|
||||
});
|
||||
it('keeps the stored selection order and does not expose private fields', async () => {
|
||||
const first = { ...instrument, uid: 'first', name: 'First' };
|
||||
const second = { ...instrument, uid: 'second', name: 'Second' };
|
||||
await seedCatalog('SBER', [
|
||||
{ uid: 'second', payload: second },
|
||||
{ uid: 'first', payload: first },
|
||||
]);
|
||||
const result = await service.read('owner', query);
|
||||
assert(result.data);
|
||||
expect(result.data.map((item) => item.uid)).toEqual(['second', 'first']);
|
||||
expect(result.data[0]).not.toHaveProperty('accountId');
|
||||
});
|
||||
});
|
||||
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 };
|
||||
}
|
||||
|
||||
Reference in New Issue
Block a user