353 lines
11 KiB
TypeScript
353 lines
11 KiB
TypeScript
import { ConfigService } from '@nestjs/config';
|
|
|
|
import { BrokerageOperationsRepository } from '../src/modules/investments/operations/infrastructure/brokerage-operations.repository';
|
|
import { PrismaService } from '../src/prisma/prisma.service';
|
|
|
|
// Real transactions are the boundary: staged pages never become financial
|
|
// results, and a failed multi-account publication must preserve every old row.
|
|
describe('Operations staging and reconciliation (isolated SQLite)', () => {
|
|
const prisma = new PrismaService(
|
|
new ConfigService({ DATABASE_URL: process.env.DATABASE_URL }),
|
|
);
|
|
const repository = new BrokerageOperationsRepository(prisma);
|
|
const attempt = {
|
|
userId: 'owner',
|
|
revision: 'revision',
|
|
id: 'work',
|
|
attemptId: 'attempt',
|
|
};
|
|
const interval = {
|
|
accountId: 'a',
|
|
from: new Date('2026-01-01Z'),
|
|
to: new Date('2026-02-01Z'),
|
|
};
|
|
const second = { ...interval, accountId: 'b' };
|
|
const operation = (id: string, value = '10') => ({
|
|
id,
|
|
date: { seconds: '1768000000' },
|
|
type: 'OPERATION_TYPE_COUPON',
|
|
payment: { units: value, nano: 1, currency: 'rub' },
|
|
});
|
|
beforeAll(async () => {
|
|
await prisma.$connect();
|
|
await prisma.user.create({
|
|
data: {
|
|
id: 'owner',
|
|
email: 'owner@example.invalid',
|
|
passwordHash: 'synthetic',
|
|
},
|
|
});
|
|
});
|
|
beforeEach(async () => {
|
|
await prisma.tbankConnection.deleteMany();
|
|
await prisma.tbankConnection.create({
|
|
data: {
|
|
userId: 'owner',
|
|
revision: 'revision',
|
|
encryptedToken: 'synthetic',
|
|
connectedAt: new Date(),
|
|
membershipAt: new Date(),
|
|
accounts: {
|
|
create: ['a', 'b'].map((accountId, upstreamOrder) => ({
|
|
accountId,
|
|
upstreamOrder,
|
|
name: accountId,
|
|
type: 'BROKERAGE',
|
|
})),
|
|
},
|
|
},
|
|
});
|
|
await prisma.tbankSyncWork.create({
|
|
data: {
|
|
...attempt,
|
|
scopeKey: 'scope',
|
|
intent: {},
|
|
progress: {},
|
|
state: 'running',
|
|
processRunId: 'run',
|
|
priority: 1,
|
|
},
|
|
});
|
|
});
|
|
afterAll(async () => prisma.$disconnect());
|
|
|
|
it('exposes only old or new financial groups to a separate SQLite reader and survives reopen', async () => {
|
|
const reader = new PrismaService(
|
|
new ConfigService({ DATABASE_URL: process.env.DATABASE_URL }),
|
|
);
|
|
await reader.$connect();
|
|
try {
|
|
for (const scope of [interval, second]) {
|
|
await repository.stagePage(attempt, scope, undefined, {
|
|
hasNext: false,
|
|
items: [operation('row', '10')],
|
|
});
|
|
}
|
|
await repository.publish(attempt, [interval, second], new Date(1000));
|
|
for (const scope of [interval, second]) {
|
|
await repository.stagePage(attempt, scope, undefined, {
|
|
hasNext: false,
|
|
items: [operation('row', '20')],
|
|
});
|
|
}
|
|
const snapshots: string[][] = [];
|
|
await Promise.all([
|
|
repository.publish(attempt, [interval, second], new Date(2000)),
|
|
(async () => {
|
|
for (let index = 0; index < 20; index += 1) {
|
|
const rows = await reader.tbankOperation.findMany({
|
|
orderBy: { accountId: 'asc' },
|
|
});
|
|
snapshots.push(
|
|
rows.map(
|
|
(row) =>
|
|
(row.payload as { payment: { units: string } }).payment.units,
|
|
),
|
|
);
|
|
}
|
|
})(),
|
|
]);
|
|
for (const snapshot of snapshots) {
|
|
expect([
|
|
['10', '10'],
|
|
['20', '20'],
|
|
]).toContainEqual(snapshot);
|
|
}
|
|
await reader.$disconnect();
|
|
await reader.$connect();
|
|
expect(
|
|
(await reader.tbankOperation.findMany()).map(
|
|
(row) =>
|
|
(row.payload as { payment: { units: string } }).payment.units,
|
|
),
|
|
).toEqual(['20', '20']);
|
|
} finally {
|
|
await reader.$disconnect();
|
|
}
|
|
});
|
|
|
|
it('limits instrument reconciliation and refuses conflicting versions across a publication group', async () => {
|
|
const instrumentOperation = {
|
|
...operation('instrument'),
|
|
instrumentUid: 'uid',
|
|
};
|
|
await repository.stagePage(attempt, interval, undefined, {
|
|
hasNext: false,
|
|
items: [instrumentOperation, operation('unrelated')],
|
|
});
|
|
await repository.publish(attempt, [interval], new Date(1000));
|
|
const subset = { ...interval, instrumentUid: 'uid' };
|
|
await repository.stagePage(attempt, subset, undefined, {
|
|
hasNext: false,
|
|
items: [],
|
|
});
|
|
await repository.publish(attempt, [subset], new Date(2000));
|
|
expect(
|
|
(await prisma.tbankOperation.findMany()).map(
|
|
(row) => row.brokerOperationId,
|
|
),
|
|
).toEqual(['unrelated']);
|
|
expect(
|
|
(
|
|
await prisma.tbankOperationCoverage.findFirstOrThrow({
|
|
where: { instrumentPredicate: '' },
|
|
})
|
|
).fetchedAt,
|
|
).toEqual(new Date(1000));
|
|
const february = {
|
|
...interval,
|
|
from: interval.to,
|
|
to: new Date('2026-03-01Z'),
|
|
};
|
|
await repository.stagePage(attempt, interval, undefined, {
|
|
hasNext: false,
|
|
items: [operation('moved')],
|
|
});
|
|
await repository.stagePage(attempt, february, undefined, {
|
|
hasNext: false,
|
|
items: [
|
|
{
|
|
...operation('moved'),
|
|
date: { seconds: String(february.from.valueOf() / 1000) },
|
|
},
|
|
],
|
|
});
|
|
await expect(
|
|
repository.publish(attempt, [interval, february], new Date(3000)),
|
|
).rejects.toThrow('Conflicting');
|
|
expect(
|
|
(await prisma.tbankOperation.findMany()).map(
|
|
(row) => row.brokerOperationId,
|
|
),
|
|
).toEqual(['unrelated']);
|
|
expect(await prisma.tbankOperationCoverage.count()).toBe(2);
|
|
});
|
|
|
|
it('keeps pages private, replaces the full group atomically, and changes revision only for row changes', async () => {
|
|
for (const scope of [interval, second]) {
|
|
await repository.stagePage(attempt, scope, undefined, {
|
|
hasNext: false,
|
|
items: [operation('same-id')],
|
|
});
|
|
}
|
|
await repository.publish(attempt, [interval, second], new Date(1000));
|
|
expect(await prisma.tbankOperation.count()).toBe(2);
|
|
expect(
|
|
(await prisma.tbankAccount.findMany()).map(
|
|
(row) => row.operationsRevision,
|
|
),
|
|
).toEqual([1, 1]);
|
|
await repository.stagePage(attempt, interval, undefined, {
|
|
hasNext: false,
|
|
items: [operation('same-id', '-20')],
|
|
});
|
|
await expect(
|
|
repository.publish(attempt, [interval, second], new Date(2000)),
|
|
).rejects.toThrow('Incomplete');
|
|
expect(
|
|
(await prisma.tbankOperation.findMany()).every((row) =>
|
|
JSON.stringify(row.payload).includes('"10"'),
|
|
),
|
|
).toBe(true);
|
|
expect(
|
|
(await prisma.tbankOperationCoverage.findMany()).every(
|
|
(row) => row.fetchedAt.valueOf() === 1000,
|
|
),
|
|
).toBe(true);
|
|
await repository.stagePage(attempt, second, undefined, {
|
|
hasNext: false,
|
|
items: [],
|
|
});
|
|
await repository.publish(attempt, [interval, second], new Date(3000));
|
|
expect(await prisma.tbankOperation.count()).toBe(1);
|
|
expect(
|
|
(await prisma.tbankAccount.findMany()).map(
|
|
(row) => row.operationsRevision,
|
|
),
|
|
).toEqual([2, 2]);
|
|
await repository.stagePage(attempt, interval, undefined, {
|
|
hasNext: false,
|
|
items: [operation('same-id', '-20')],
|
|
});
|
|
await repository.publish(attempt, [interval], new Date(4000));
|
|
expect(
|
|
(
|
|
await prisma.tbankAccount.findFirstOrThrow({
|
|
where: { accountId: 'a' },
|
|
})
|
|
).operationsRevision,
|
|
).toBe(2);
|
|
expect(
|
|
(
|
|
await prisma.tbankOperationCoverage.findFirstOrThrow({
|
|
where: { accountId: 'a' },
|
|
})
|
|
).fetchedAt,
|
|
).toEqual(new Date(4000));
|
|
});
|
|
|
|
it('reconciles a changed date and removes cancelled operations without changing another account', async () => {
|
|
await repository.stagePage(attempt, interval, undefined, {
|
|
hasNext: false,
|
|
items: [operation('moved'), operation('cancelled')],
|
|
});
|
|
await repository.stagePage(attempt, second, undefined, {
|
|
hasNext: false,
|
|
items: [operation('cancelled')],
|
|
});
|
|
await repository.publish(attempt, [interval, second], new Date(1000));
|
|
await expect(
|
|
repository.stagePage(attempt, interval, undefined, {
|
|
hasNext: false,
|
|
items: [
|
|
operation('cancelled'),
|
|
{ ...operation('cancelled'), state: 'OPERATION_STATE_CANCELED' },
|
|
],
|
|
}),
|
|
).rejects.toThrow('Conflicting');
|
|
const moved = {
|
|
...operation('moved', '-123'),
|
|
date: { seconds: '1769000000' },
|
|
state: 'OPERATION_STATE_EXECUTED',
|
|
};
|
|
await repository.stagePage(attempt, interval, undefined, {
|
|
hasNext: false,
|
|
items: [
|
|
moved,
|
|
{ ...operation('cancelled'), state: 'OPERATION_STATE_CANCELED' },
|
|
],
|
|
});
|
|
expect(await prisma.tbankOperation.count()).toBe(3);
|
|
await repository.publish(attempt, [interval], new Date(2000));
|
|
const rows = await prisma.tbankOperation.findMany({
|
|
orderBy: { accountId: 'asc' },
|
|
});
|
|
expect(rows).toHaveLength(2);
|
|
expect(rows[0].payload).toEqual(moved);
|
|
expect(rows[0].occurredAt).toEqual(new Date(1769000000000));
|
|
expect(rows[1].accountId).toBe('b');
|
|
expect(rows[1].brokerOperationId).toBe('cancelled');
|
|
});
|
|
|
|
it('rejects malformed pagination and conflicting identities, filters exact boundaries, and fences superseded work', async () => {
|
|
await expect(
|
|
repository.stagePage(attempt, interval, undefined, {
|
|
hasNext: true,
|
|
items: [],
|
|
}),
|
|
).rejects.toThrow('continuation');
|
|
await expect(
|
|
repository.stagePage(attempt, interval, undefined, {
|
|
hasNext: false,
|
|
items: [{ ...operation(''), cursor: 'not-an-id' }],
|
|
}),
|
|
).rejects.toThrow('identity');
|
|
expect(await prisma.tbankSyncStage.count()).toBe(0);
|
|
const edge = {
|
|
...operation('edge'),
|
|
date: { seconds: String(interval.to.valueOf() / 1000) },
|
|
};
|
|
await repository.stagePage(attempt, interval, undefined, {
|
|
hasNext: true,
|
|
nextCursor: 'next',
|
|
items: [operation('id'), operation('id'), edge],
|
|
});
|
|
expect(await prisma.tbankOperation.count()).toBe(0);
|
|
await expect(
|
|
repository.stagePage(attempt, interval, 'next', {
|
|
hasNext: true,
|
|
nextCursor: 'next',
|
|
items: [],
|
|
}),
|
|
).rejects.toThrow('continuation');
|
|
await expect(
|
|
repository.stagePage(attempt, interval, 'next', {
|
|
hasNext: false,
|
|
items: [operation('id', '11')],
|
|
}),
|
|
).rejects.toThrow('Conflicting');
|
|
await repository.stagePage(attempt, interval, 'next', {
|
|
hasNext: false,
|
|
items: [operation('id')],
|
|
});
|
|
await repository.publish(attempt, [interval], new Date());
|
|
expect(
|
|
(await prisma.tbankOperation.findMany()).map(
|
|
(row) => row.brokerOperationId,
|
|
),
|
|
).toEqual(['id']);
|
|
await prisma.tbankSyncWork.update({
|
|
where: { id: attempt.id },
|
|
data: { attemptId: 'replacement' },
|
|
});
|
|
expect(
|
|
await repository.stagePage(attempt, second, undefined, {
|
|
hasNext: false,
|
|
items: [],
|
|
}),
|
|
).toBe(false);
|
|
expect(await repository.publish(attempt, [second], new Date())).toBe(false);
|
|
expect(await prisma.tbankOperation.count()).toBe(1);
|
|
});
|
|
});
|