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); }); });