453 lines
18 KiB
TypeScript
453 lines
18 KiB
TypeScript
import type { Server } from 'node:http';
|
|
|
|
import { INestApplication } from '@nestjs/common';
|
|
import { JwtService } from '@nestjs/jwt';
|
|
import { Test } from '@nestjs/testing';
|
|
import assert from 'node:assert/strict';
|
|
import { setTimeout as delay } from 'node:timers/promises';
|
|
import request from 'supertest';
|
|
|
|
import type { IncomeAnalyticsResourceDto } from '../src/modules/investments/income/income-analytics.dto';
|
|
|
|
import { AppModule } from '../src/app.module';
|
|
import { InstrumentMetadataRepository } from '../src/integrations/tbank/instrument-metadata.repository';
|
|
import { TbankClient } from '../src/integrations/tbank/tbank.client';
|
|
import { configureApp } from '../src/main';
|
|
import { BrokerageOperationsRepository } from '../src/modules/investments/operations/infrastructure/brokerage-operations.repository';
|
|
import { BrokerageOperationsResourceDto } from '../src/modules/investments/operations/presentation/brokerage-operation.dto';
|
|
import { TbankHourlyScheduler } from '../src/modules/investments/tbank-sync/tbank-hourly.scheduler';
|
|
import { PrismaService } from '../src/prisma/prisma.service';
|
|
|
|
const uid = '11111111-1111-4111-8111-111111111111';
|
|
const period = 'from=2025-01-01&to=2025-01-31';
|
|
describe('persistent ledger HTTP and cursor (SQLite)', () => {
|
|
let app: INestApplication<Server>;
|
|
let prisma: PrismaService;
|
|
const upstream = {
|
|
validateToken: jest.fn(),
|
|
getOpenAccounts: jest.fn(),
|
|
getOperationsByCursor: jest.fn(),
|
|
};
|
|
const auth = (owner = 'owner') =>
|
|
`Bearer ${new JwtService().sign({ sub: owner, type: 'access' }, { secret: process.env.JWT_SECRET })}`;
|
|
const get = (query = '', account = 'all', owner = 'owner') =>
|
|
request(app.getHttpServer())
|
|
.get(
|
|
`/api/v1/brokerage-accounts/${account}/operations?${period}&${query}`,
|
|
)
|
|
.set('Authorization', auth(owner))
|
|
.timeout(30_000);
|
|
const put = (owner = 'owner') =>
|
|
request(app.getHttpServer())
|
|
.put('/api/v1/tbank-connection')
|
|
.set('Authorization', auth(owner))
|
|
.send({ token: 'synthetic' });
|
|
beforeAll(async () => {
|
|
const module = await Test.createTestingModule({ imports: [AppModule] })
|
|
.overrideProvider(TbankClient)
|
|
.useValue(upstream)
|
|
.overrideProvider(TbankHourlyScheduler)
|
|
.useValue({ tick: jest.fn() })
|
|
.compile();
|
|
app = module.createNestApplication();
|
|
configureApp(app);
|
|
await app.init();
|
|
await app.listen(0, '127.0.0.1');
|
|
prisma = app.get(PrismaService);
|
|
for (const id of ['owner', 'other']) {
|
|
await prisma.user.create({
|
|
data: { id, email: `${id}@example.invalid`, passwordHash: 'synthetic' },
|
|
});
|
|
}
|
|
});
|
|
afterAll(async () => app?.close());
|
|
beforeEach(async () => {
|
|
await prisma.tbankConnection.deleteMany();
|
|
jest.clearAllMocks();
|
|
upstream.validateToken.mockResolvedValue(
|
|
['a', 'b'].map((id) => ({
|
|
id,
|
|
name: id,
|
|
type: 'ACCOUNT_TYPE_TINKOFF',
|
|
openedDate: { seconds: '1577836800' },
|
|
})),
|
|
);
|
|
upstream.getOperationsByCursor
|
|
.mockReset()
|
|
.mockResolvedValue({ hasNext: false, items: [] });
|
|
await put().expect(200);
|
|
const { revision } = await prisma.tbankConnection.findUniqueOrThrow({
|
|
where: { userId: 'owner' },
|
|
});
|
|
for (const accountId of ['a', 'b']) {
|
|
await prisma.tbankOperationCoverage.create({
|
|
data: {
|
|
userId: 'owner',
|
|
revision,
|
|
accountId,
|
|
from: new Date('2020-01-01Z'),
|
|
to: new Date(),
|
|
asOf: new Date(),
|
|
fetchedAt: new Date(),
|
|
},
|
|
});
|
|
for (const id of ['1', '2', '3']) {
|
|
await prisma.tbankOperation.create({
|
|
data: {
|
|
userId: 'owner',
|
|
revision,
|
|
accountId,
|
|
brokerOperationId: id,
|
|
occurredAt: new Date('2025-01-02Z'),
|
|
category: id === '3' ? 'trade' : 'income',
|
|
instrumentUid: uid,
|
|
payload: {
|
|
id,
|
|
instrumentUid: uid,
|
|
date: { seconds: '1735776000' },
|
|
type: id === '3' ? 'OPERATION_TYPE_BUY' : 'OPERATION_TYPE_COUPON',
|
|
payment: { currency: 'rub', units: '1', nano: 1 },
|
|
},
|
|
},
|
|
});
|
|
}
|
|
}
|
|
});
|
|
it('reads ordered pages with identical IDs/timestamps across accounts and no upstream RPC, including after reopen', async () => {
|
|
let cursor: null | string = null;
|
|
const ids: string[] = [];
|
|
do {
|
|
const response = await get(
|
|
`limit=2${cursor ? `&cursor=${encodeURIComponent(cursor)}` : ''}`,
|
|
).expect(200);
|
|
const result = response.body as BrokerageOperationsResourceDto;
|
|
expect(result.meta.availability).toBe('available');
|
|
expect(result.meta.completeness).toBe('complete');
|
|
assert(result.data);
|
|
ids.push(
|
|
...result.data.items.map((item) => `${item.account?.id}/${item.id}`),
|
|
);
|
|
expect(result.data.items[0].payment).toEqual({
|
|
currency: 'RUB',
|
|
value: '1.000000001',
|
|
});
|
|
cursor = result.data.nextCursor;
|
|
if (cursor) await prisma.$disconnect();
|
|
} while (cursor);
|
|
expect(ids).toEqual(['a/1', 'a/2', 'a/3', 'b/1', 'b/2', 'b/3']);
|
|
expect(upstream.getOperationsByCursor).not.toHaveBeenCalled();
|
|
expect(upstream.getOpenAccounts).not.toHaveBeenCalled();
|
|
});
|
|
it('uses the persisted subsecond timestamp for filtering, cursor order and income entries', async () => {
|
|
await prisma.tbankOperation.deleteMany();
|
|
const { revision } = await prisma.tbankConnection.findUniqueOrThrow({
|
|
where: { userId: 'owner' },
|
|
});
|
|
const repository = app.get(BrokerageOperationsRepository);
|
|
const interval = {
|
|
accountId: 'a',
|
|
from: new Date('2025-01-02T00:00:00.500Z'),
|
|
to: new Date('2025-01-02T00:00:01.000Z'),
|
|
};
|
|
const reserved = await repository.request(
|
|
'owner',
|
|
revision,
|
|
[interval],
|
|
'subsecond-test',
|
|
true,
|
|
new Set(),
|
|
1,
|
|
);
|
|
assert(reserved);
|
|
const attempt = {
|
|
userId: 'owner',
|
|
revision,
|
|
id: reserved.work.id,
|
|
attemptId: reserved.work.attemptId,
|
|
};
|
|
await expect(
|
|
repository.stagePage(attempt, interval, undefined, {
|
|
hasNext: false,
|
|
items: [{ id: 'invalid', date: {} }],
|
|
}),
|
|
).rejects.toThrow('Invalid operation timestamp');
|
|
await repository.stagePage(attempt, interval, undefined, {
|
|
hasNext: false,
|
|
items: [400, 600, 800].map((milliseconds) => ({
|
|
id: String(milliseconds),
|
|
date: { seconds: '1735776000', nanos: milliseconds * 1_000_000 },
|
|
type: 'OPERATION_TYPE_COUPON',
|
|
payment: { currency: 'rub', units: '1', nano: 1 },
|
|
})),
|
|
});
|
|
await repository.publish(attempt, [interval], new Date(), true);
|
|
const saved = await prisma.tbankOperation.findMany({
|
|
orderBy: { occurredAt: 'desc' },
|
|
});
|
|
const expected = [
|
|
{ id: '800', occurredAt: '2025-01-02T00:00:00.800Z' },
|
|
{ id: '600', occurredAt: '2025-01-02T00:00:00.600Z' },
|
|
];
|
|
expect(
|
|
saved.map((row) => ({
|
|
id: row.brokerOperationId,
|
|
occurredAt: row.occurredAt.toISOString(),
|
|
})),
|
|
).toEqual(expected);
|
|
const items: { id: string; occurredAt: string }[] = [];
|
|
let cursor: null | string = null;
|
|
do {
|
|
const response = await request(app.getHttpServer())
|
|
.get(
|
|
`/api/v1/brokerage-accounts/a/operations?from=${interval.from.toISOString()}&to=2025-01-02T00:00:00.900Z&limit=1&sync=observe${cursor ? `&cursor=${encodeURIComponent(cursor)}` : ''}`,
|
|
)
|
|
.set('Authorization', auth())
|
|
.expect(200);
|
|
const result = response.body as BrokerageOperationsResourceDto;
|
|
assert(result.data);
|
|
expect(result.data.items).toHaveLength(1);
|
|
items.push(
|
|
...result.data.items.map(({ id, occurredAt }) => ({ id, occurredAt })),
|
|
);
|
|
cursor = result.data.nextCursor;
|
|
} while (cursor);
|
|
const income = (
|
|
await request(app.getHttpServer())
|
|
.get(
|
|
`/api/v1/brokerage-accounts/a/income-analytics?${period}&sync=observe`,
|
|
)
|
|
.set('Authorization', auth())
|
|
.expect(200)
|
|
).body as IncomeAnalyticsResourceDto;
|
|
expect(items).toEqual(expected);
|
|
expect(
|
|
income.data?.entries.map(({ id, occurredAt }) => ({ id, occurredAt })),
|
|
).toEqual(expected);
|
|
expect(income.data?.totalsByCurrency[0].total).toEqual({
|
|
currency: 'RUB',
|
|
value: '2.000000002',
|
|
});
|
|
expect(upstream.getOperationsByCursor).not.toHaveBeenCalled();
|
|
});
|
|
it.each(['default', 'today', 'today timestamp'])(
|
|
'keeps observed current coverage complete through %s period pagination without RPC',
|
|
async (periodKind) => {
|
|
const observedAt = new Date(Date.now() - 1000);
|
|
await prisma.tbankOperationCoverage.updateMany({
|
|
data: {
|
|
to: observedAt,
|
|
asOf: observedAt,
|
|
fetchedAt: observedAt,
|
|
},
|
|
});
|
|
await prisma.tbankOperation.updateMany({
|
|
data: { occurredAt: new Date(observedAt.valueOf() - 1000) },
|
|
});
|
|
const query =
|
|
periodKind === 'default'
|
|
? ''
|
|
: `&to=${observedAt.toISOString().slice(0, 10)}${periodKind === 'today timestamp' ? 'T23:59:59.999Z' : ''}`;
|
|
const page = (cursor?: string, observe = false) =>
|
|
request(app.getHttpServer())
|
|
.get(
|
|
`/api/v1/brokerage-accounts/all/operations?${observe ? 'sync=observe&' : ''}limit=1${query}${cursor ? `&cursor=${encodeURIComponent(cursor)}` : ''}`,
|
|
)
|
|
.set('Authorization', auth());
|
|
const first = (await page().expect(200))
|
|
.body as BrokerageOperationsResourceDto;
|
|
expect(first.meta.availability).toBe('available');
|
|
expect(first.meta.completeness).toBe('complete');
|
|
assert(first.data?.nextCursor);
|
|
const second = (await page(first.data.nextCursor).expect(200))
|
|
.body as BrokerageOperationsResourceDto;
|
|
expect(second.meta.availability).toBe('available');
|
|
expect(second.meta.completeness).toBe('complete');
|
|
expect(second.meta.coverage?.missingRanges).toEqual([]);
|
|
const observed = (await page(first.data.nextCursor, true).expect(200))
|
|
.body as BrokerageOperationsResourceDto;
|
|
expect(observed.meta.availability).toBe('available');
|
|
expect(observed.meta.completeness).toBe('complete');
|
|
expect(upstream.getOperationsByCursor).not.toHaveBeenCalled();
|
|
},
|
|
);
|
|
it('keeps data revisions stable while observational reads expose current coverage', async () => {
|
|
const observe = () =>
|
|
request(app.getHttpServer())
|
|
.get('/api/v1/brokerage-accounts/all/operations?sync=observe&limit=100')
|
|
.set('Authorization', auth());
|
|
const first = (await observe().expect(200))
|
|
.body as BrokerageOperationsResourceDto;
|
|
await delay(20);
|
|
const second = (await observe().expect(200))
|
|
.body as BrokerageOperationsResourceDto;
|
|
expect(second.meta.coverage?.requested).not.toEqual(
|
|
first.meta.coverage?.requested,
|
|
);
|
|
expect(second.meta.dataRevision).toBe(first.meta.dataRevision);
|
|
await prisma.tbankOperationCoverage.updateMany({
|
|
data: {
|
|
to: new Date(Date.now() - 60_000),
|
|
asOf: new Date(Date.now() - 60_000),
|
|
},
|
|
});
|
|
const changed = (await observe().expect(200))
|
|
.body as BrokerageOperationsResourceDto;
|
|
expect(changed.meta.coverage?.asOf).not.toBe(second.meta.coverage?.asOf);
|
|
expect(upstream.getOperationsByCursor).not.toHaveBeenCalled();
|
|
});
|
|
it('reads shared metadata from SQLite without invalidating a row cursor', async () => {
|
|
const before = (await get('limit=1&sync=observe').expect(200))
|
|
.body as BrokerageOperationsResourceDto;
|
|
assert(before.data?.nextCursor);
|
|
const { revision } = await prisma.tbankConnection.findUniqueOrThrow({
|
|
where: { userId: 'owner' },
|
|
});
|
|
const metadata = new InstrumentMetadataRepository(prisma);
|
|
const reservation = await metadata.reserve(
|
|
{ userId: 'owner', revision },
|
|
{ instrumentUid: uid },
|
|
'metadata-test',
|
|
new Set(),
|
|
);
|
|
assert(reservation);
|
|
await metadata.publish(
|
|
{
|
|
userId: 'owner',
|
|
revision,
|
|
id: reservation.work.id,
|
|
attemptId: reservation.work.attemptId,
|
|
},
|
|
[
|
|
{
|
|
type: 'uid',
|
|
id: uid,
|
|
data: { uid, name: 'Synthetic', brand: { logoName: 'saved.png' } },
|
|
},
|
|
],
|
|
new Date(),
|
|
);
|
|
await prisma.$disconnect();
|
|
const after = (await get('limit=1&sync=observe').expect(200))
|
|
.body as BrokerageOperationsResourceDto;
|
|
expect(after.data?.items[0].logoName).toBe('saved.png');
|
|
expect(after.meta.dataRevision).toBe(before.meta.dataRevision);
|
|
await get(
|
|
`limit=1&sync=observe&cursor=${encodeURIComponent(before.data.nextCursor)}`,
|
|
).expect(200);
|
|
expect(upstream.getOperationsByCursor).not.toHaveBeenCalled();
|
|
});
|
|
it('binds a signed cursor to equivalent filters, user, membership, row revision and connection; status alone does not break it', async () => {
|
|
const first = (
|
|
await get('limit=1&categories=trade,income&sync=observe').expect(200)
|
|
).body as BrokerageOperationsResourceDto;
|
|
assert(first.data?.nextCursor);
|
|
const cursor = encodeURIComponent(first.data.nextCursor);
|
|
const next = `limit=1&categories=income,trade,income&sync=observe&cursor=${cursor}`;
|
|
await prisma.tbankOperationCoverage.updateMany({
|
|
data: { fetchedAt: new Date(0) },
|
|
});
|
|
await get(next).expect(200);
|
|
await get(next.replace('limit=1', 'limit=2')).expect(400);
|
|
await get(`${next}x`).expect(400);
|
|
await put('other').expect(200);
|
|
await get(next, 'all', 'other').expect(400);
|
|
await prisma.tbankAccount.updateMany({
|
|
where: { userId: 'owner', accountId: 'b' },
|
|
data: { operationsRevision: { increment: 1 } },
|
|
});
|
|
await get(next).expect(400);
|
|
const single = (await get('limit=1&sync=observe', 'a').expect(200))
|
|
.body as BrokerageOperationsResourceDto;
|
|
assert(single.data?.nextCursor);
|
|
const singleNext = `limit=1&sync=observe&cursor=${encodeURIComponent(single.data.nextCursor)}`;
|
|
await prisma.tbankConnection.update({
|
|
where: { userId: 'owner' },
|
|
data: { membershipRevision: { increment: 1 } },
|
|
});
|
|
await get(singleNext, 'a').expect(200);
|
|
await put().expect(200);
|
|
await get(singleNext, 'a').expect(400);
|
|
});
|
|
it('does not confuse pagination/categories/instrument subset with interval completeness', async () => {
|
|
await prisma.tbankOperationCoverage.deleteMany();
|
|
const { revision } = await prisma.tbankConnection.findUniqueOrThrow({
|
|
where: { userId: 'owner' },
|
|
});
|
|
await prisma.tbankOperationCoverage.create({
|
|
data: {
|
|
userId: 'owner',
|
|
revision,
|
|
accountId: 'a',
|
|
instrumentPredicate: uid,
|
|
from: new Date('2025-01-01Z'),
|
|
to: new Date('2025-01-04Z'),
|
|
asOf: new Date('2025-01-04Z'),
|
|
fetchedAt: new Date(),
|
|
},
|
|
});
|
|
const partial = (
|
|
await get(
|
|
`instrumentUid=${uid}&categories=income&limit=100&sync=observe`,
|
|
'a',
|
|
).expect(200)
|
|
).body as BrokerageOperationsResourceDto;
|
|
expect(partial.meta.availability).toBe('initial-loading');
|
|
expect(partial.meta.completeness).toBe('incomplete');
|
|
expect(partial.data).toBeNull();
|
|
expect(partial.meta.coverage?.missingRanges.length).toBeGreaterThan(0);
|
|
const whole = (await get('sync=observe', 'a').expect(200))
|
|
.body as BrokerageOperationsResourceDto;
|
|
expect(whole.meta.availability).toBe('available');
|
|
expect(whole.meta.completeness).toBe('complete');
|
|
expect(whole.data?.items).toEqual([]);
|
|
expect(upstream.getOperationsByCursor.mock.calls.length).toBeGreaterThan(0);
|
|
});
|
|
it('keeps an unknown history boundary explicit even when the shown interval is complete', async () => {
|
|
await prisma.tbankAccount.updateMany({ data: { openedOn: null } });
|
|
await prisma.tbankOperationCoverage.deleteMany({ where: { accountId: 'a' } });
|
|
await prisma.tbankSyncWork.deleteMany({ where: { userId: 'owner' } });
|
|
const result = (await get('sync=observe', 'a').expect(200))
|
|
.body as BrokerageOperationsResourceDto;
|
|
expect(result.meta.coverage?.historyBoundary).toBe('unknown');
|
|
expect(result.meta.coverage?.historyState).toBe('loading');
|
|
}, 30_000);
|
|
|
|
it('returns cold missing while upstream is blocked, then confirms an empty interval; observe never retries failure', async () => {
|
|
await prisma.tbankOperation.deleteMany();
|
|
await prisma.tbankOperationCoverage.deleteMany();
|
|
let unblock: (value: { hasNext: boolean; items: never[] }) => void = () => {
|
|
throw new Error('not initialized');
|
|
};
|
|
upstream.getOperationsByCursor
|
|
.mockReturnValueOnce(
|
|
new Promise((resolve) => {
|
|
unblock = resolve;
|
|
}),
|
|
)
|
|
.mockRejectedValue(new Error('synthetic outage'));
|
|
const cold = (await get('', 'a').expect(200))
|
|
.body as BrokerageOperationsResourceDto;
|
|
expect(cold.data).toBeNull();
|
|
expect(cold.meta.availability).toBe('initial-loading');
|
|
expect(cold.meta.completeness).toBe('incomplete');
|
|
await get('sync=observe', 'a').expect(200);
|
|
expect(upstream.getOperationsByCursor).toHaveBeenCalledTimes(1);
|
|
// eslint-disable-next-line sonarjs/no-extra-arguments -- The typed callback is assigned the promise resolver before invocation.
|
|
unblock({ hasNext: false, items: [] });
|
|
for (let index = 0; index < 300; index += 1) {
|
|
if (!(await prisma.tbankSyncWork.count({ where: { state: 'running' } }))) {
|
|
break;
|
|
}
|
|
await delay(5);
|
|
}
|
|
const count = upstream.getOperationsByCursor.mock.calls.length;
|
|
const result = (await get('sync=observe', 'a').expect(200))
|
|
.body as BrokerageOperationsResourceDto;
|
|
expect(result.data?.items).toEqual([]);
|
|
expect(result.meta.availability).toBe('available');
|
|
expect(result.meta.completeness).toBe('complete');
|
|
await get('sync=observe', 'a').expect(200);
|
|
expect(upstream.getOperationsByCursor.mock.calls.length).toBeGreaterThanOrEqual(count);
|
|
});
|
|
});
|