658 lines
24 KiB
TypeScript
658 lines
24 KiB
TypeScript
import { ConfigService } from '@nestjs/config';
|
|
import assert from 'node:assert/strict';
|
|
import { randomUUID } from 'node:crypto';
|
|
import { setImmediate as immediate } from 'node:timers/promises';
|
|
|
|
import { TbankAccessService } from '../src/integrations/tbank/tbank-access.service';
|
|
import { TbankTokenService } from '../src/integrations/tbank/tbank-token.service';
|
|
import { CatalogEnrichmentService } from '../src/modules/investments/instruments/application/catalog-enrichment.service';
|
|
import { CatalogDiscoveryRepository } from '../src/modules/investments/instruments/infrastructure/catalog-discovery.repository';
|
|
import { BrokerageOperationsSyncService } from '../src/modules/investments/operations/application/brokerage-operations-sync.service';
|
|
import { BrokerageOperationsRepository } from '../src/modules/investments/operations/infrastructure/brokerage-operations.repository';
|
|
import { TbankHourlyScheduler } from '../src/modules/investments/tbank-sync/tbank-hourly.scheduler';
|
|
import { TbankSyncRepository } from '../src/modules/investments/tbank-sync/tbank-sync.repository';
|
|
import { PrismaService } from '../src/prisma/prisma.service';
|
|
|
|
type HourlyProgressView = {
|
|
result?: string;
|
|
sections: { accountId?: string; errorCode: null | string; section: string }[];
|
|
};
|
|
|
|
// Real SQLite proves reservation, outcome derivation, recovery and fences;
|
|
// the step owners are controlled so each failure mode can be injected.
|
|
describe('hourly attempt lifecycle (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 repository = new TbankSyncRepository(prisma);
|
|
const access = { local: jest.fn() };
|
|
const membership = { request: jest.fn() };
|
|
const portfolio = { request: jest.fn() };
|
|
const operations = { request: jest.fn() };
|
|
const events = { refresh: jest.fn(), get: jest.fn() };
|
|
const logger = { event: jest.fn() };
|
|
const createScheduler = () =>
|
|
new TbankHourlyScheduler(
|
|
prisma,
|
|
repository,
|
|
access as never,
|
|
membership as never,
|
|
portfolio as never,
|
|
operations as never,
|
|
events as never,
|
|
logger as never,
|
|
);
|
|
const USER = 'hourly-owner';
|
|
const REVISION = 'hourly-revision';
|
|
|
|
function child(
|
|
state: 'complete' | 'failed' | 'running',
|
|
errorCode: null | string,
|
|
scopeKey: string,
|
|
) {
|
|
return prisma.tbankSyncWork.create({
|
|
data: {
|
|
id: randomUUID(),
|
|
userId: USER,
|
|
revision: REVISION,
|
|
scopeKey: `test:${scopeKey}:${randomUUID()}`,
|
|
attemptId: randomUUID(),
|
|
processRunId: 'child-process',
|
|
priority: 0,
|
|
intent: {},
|
|
progress: {},
|
|
state,
|
|
errorCode,
|
|
},
|
|
});
|
|
}
|
|
|
|
function hourlyRow() {
|
|
return prisma.tbankSyncWork.findFirst({
|
|
where: { userId: USER, scopeKey: 'hourly' },
|
|
});
|
|
}
|
|
|
|
async function pump(turns = 20) {
|
|
for (let i = 0; i < turns; i++) {
|
|
await immediate();
|
|
await jest.advanceTimersByTimeAsync(50);
|
|
}
|
|
}
|
|
|
|
async function runWalk(scheduler: TbankHourlyScheduler, advanceMs = 50) {
|
|
await scheduler.tick();
|
|
for (let i = 0; i < 400; i++) {
|
|
await immediate();
|
|
await jest.advanceTimersByTimeAsync(advanceMs);
|
|
const row = await hourlyRow();
|
|
if (row && row.state !== 'running') return row;
|
|
}
|
|
throw new Error('hourly attempt did not settle');
|
|
}
|
|
|
|
// Sequential per-step bounds: advance past each bound until the attempt
|
|
// closes. Covers walks whose awaits hang before any durable step row exists.
|
|
async function runBoundedWalk(scheduler: TbankHourlyScheduler) {
|
|
await scheduler.tick();
|
|
for (let i = 0; i < 12; i++) {
|
|
await pump(5);
|
|
await jest.advanceTimersByTimeAsync(16 * 60 * 1000);
|
|
const row = await hourlyRow();
|
|
if (row && row.state !== 'running') return row;
|
|
}
|
|
throw new Error('hourly attempt did not settle');
|
|
}
|
|
|
|
beforeAll(async () => {
|
|
await prisma.$connect();
|
|
await prisma.user.create({
|
|
data: { id: USER, email: 'hourly@example.invalid', passwordHash: 'x' },
|
|
});
|
|
});
|
|
afterAll(async () => prisma.$disconnect());
|
|
beforeEach(async () => {
|
|
jest.resetAllMocks();
|
|
await prisma.tbankSyncWork.deleteMany();
|
|
await prisma.tbankConnection.deleteMany();
|
|
await prisma.tbankConnection.create({
|
|
data: {
|
|
userId: USER,
|
|
revision: REVISION,
|
|
encryptedToken: tokens.encrypt('synthetic'),
|
|
connectedAt: new Date(),
|
|
},
|
|
});
|
|
jest.useFakeTimers({
|
|
doNotFake: ['queueMicrotask', 'setImmediate', 'nextTick'],
|
|
});
|
|
access.local.mockResolvedValue({
|
|
revision: REVISION,
|
|
accounts: [{ accountId: 'a' }, { accountId: 'b' }],
|
|
});
|
|
membership.request.mockImplementation(async () => {
|
|
const row = await child('complete', null, 'membership');
|
|
return { id: row.id, done: Promise.resolve() };
|
|
});
|
|
portfolio.request.mockImplementation(
|
|
async (_userId: string, _revision: string, accountId: string) => {
|
|
const row = await child('complete', null, `portfolio:${accountId}`);
|
|
return { id: row.id, done: Promise.resolve() };
|
|
},
|
|
);
|
|
operations.request.mockImplementation(async () => {
|
|
const row = await child('complete', null, 'operations');
|
|
return row.id;
|
|
});
|
|
events.refresh.mockResolvedValue([]);
|
|
events.get.mockResolvedValue(undefined);
|
|
});
|
|
afterEach(() => {
|
|
jest.useRealTimers();
|
|
jest.restoreAllMocks();
|
|
});
|
|
|
|
it('settles the first walk succeeded only when every step row completed', async () => {
|
|
const row = await runWalk(createScheduler());
|
|
expect(row.state).toBe('complete');
|
|
const progress = row.progress as HourlyProgressView;
|
|
expect(progress.result).toBe('succeeded');
|
|
expect(progress.sections).toEqual([]);
|
|
const read = await repository.hourly(USER);
|
|
expect(read?.state).toBe('succeeded');
|
|
});
|
|
|
|
it('settles partial and keeps successful steps when one account fails', async () => {
|
|
portfolio.request.mockImplementation(
|
|
async (_userId: string, _revision: string, accountId: string) => {
|
|
const failed = accountId === 'b';
|
|
const row = await child(
|
|
failed ? 'failed' : 'complete',
|
|
failed ? 'TBANK_UNAVAILABLE' : null,
|
|
`portfolio:${accountId}`,
|
|
);
|
|
return { id: row.id, done: Promise.resolve() };
|
|
},
|
|
);
|
|
const row = await runWalk(createScheduler());
|
|
const progress = row.progress as HourlyProgressView;
|
|
expect(progress.result).toBe('partial');
|
|
expect(progress.sections).toHaveLength(1);
|
|
expect(progress.sections[0].section).toBe('portfolio');
|
|
expect(progress.sections[0].accountId).toBe('b');
|
|
expect(progress.sections[0].errorCode).toBe('TBANK_UNAVAILABLE');
|
|
});
|
|
|
|
it('settles failed when the membership step fails before any update', async () => {
|
|
membership.request.mockImplementation(async () => {
|
|
const row = await child('failed', 'INVALID_TBANK_TOKEN', 'membership');
|
|
return { id: row.id, done: Promise.resolve() };
|
|
});
|
|
const row = await runWalk(createScheduler());
|
|
const progress = row.progress as HourlyProgressView;
|
|
expect(progress.result).toBe('failed');
|
|
expect(row.errorCode).toBe('INVALID_TBANK_TOKEN');
|
|
expect(progress.sections[0].section).toBe('membership');
|
|
});
|
|
|
|
it('never succeeds when no required step succeeded', async () => {
|
|
membership.request.mockImplementation(async () => {
|
|
const row = await child('failed', 'TBANK_UNAVAILABLE', 'membership');
|
|
return { id: row.id, done: Promise.resolve() };
|
|
});
|
|
portfolio.request.mockImplementation(
|
|
async (_userId: string, _revision: string, accountId: string) => {
|
|
const row = await child(
|
|
'failed',
|
|
'TBANK_UNAVAILABLE',
|
|
`portfolio:${accountId}`,
|
|
);
|
|
return { id: row.id, done: Promise.resolve() };
|
|
},
|
|
);
|
|
operations.request.mockRejectedValue(new Error('db'));
|
|
events.refresh.mockRejectedValue(new Error('db'));
|
|
const row = await runWalk(createScheduler());
|
|
const progress = row.progress as HourlyProgressView;
|
|
expect(progress.result).toBe('failed');
|
|
expect(progress.result).not.toBe('succeeded');
|
|
assert.ok(progress.sections.length > 0);
|
|
});
|
|
|
|
it('caps the outcome at partial when the coordinator itself breaks', async () => {
|
|
access.local.mockRejectedValueOnce(new Error('db'));
|
|
const row = await runWalk(createScheduler());
|
|
const progress = row.progress as HourlyProgressView;
|
|
expect(progress.result).toBe('partial');
|
|
expect(row.errorCode).toBe('INTERRUPTED');
|
|
});
|
|
|
|
it('records a section whose dispatch failed as a problem', async () => {
|
|
events.refresh.mockRejectedValue(new Error('db'));
|
|
const row = await runWalk(createScheduler());
|
|
const progress = row.progress as HourlyProgressView;
|
|
expect(progress.result).toBe('partial');
|
|
const eventsSection = progress.sections.find(
|
|
(entry) => entry.section === 'events',
|
|
);
|
|
assert.ok(eventsSection);
|
|
expect(eventsSection.errorCode).toBe('TBANK_UNAVAILABLE');
|
|
expect(row.errorCode).toBe('TBANK_UNAVAILABLE');
|
|
});
|
|
|
|
it('closes a stuck step as DEADLINE and keeps its row running', async () => {
|
|
operations.request.mockImplementation(async () => {
|
|
const row = await child('running', null, 'operations');
|
|
return row.id;
|
|
});
|
|
const scheduler = createScheduler();
|
|
await scheduler.tick();
|
|
await pump();
|
|
await jest.advanceTimersByTimeAsync(16 * 60 * 1000);
|
|
await pump(20);
|
|
const row = await hourlyRow();
|
|
assert.ok(row);
|
|
expect(row.state).not.toBe('running');
|
|
const progress = row.progress as HourlyProgressView;
|
|
expect(progress.result).toBe('partial');
|
|
assert.ok(
|
|
progress.sections.some(
|
|
(entry) =>
|
|
entry.section === 'operations' && entry.errorCode === 'DEADLINE',
|
|
),
|
|
);
|
|
const stuck = await prisma.tbankSyncWork.findFirst({
|
|
where: { userId: USER, state: 'running' },
|
|
});
|
|
assert.ok(stuck, 'the wait bound does not cancel the upstream call');
|
|
});
|
|
|
|
it('settles the attempt when a portfolio dispatch never resolves', async () => {
|
|
portfolio.request.mockImplementation(() => new Promise(() => undefined));
|
|
const row = await runBoundedWalk(createScheduler());
|
|
const progress = row.progress as HourlyProgressView;
|
|
expect(progress.result).toBe('partial');
|
|
const problems = progress.sections.filter(
|
|
(entry) => entry.section === 'portfolio',
|
|
);
|
|
expect(problems).toHaveLength(2);
|
|
expect(problems[0].errorCode).toBe('TBANK_UNAVAILABLE');
|
|
});
|
|
|
|
it('settles the attempt when the membership dispatch never resolves', async () => {
|
|
membership.request.mockImplementation(() => new Promise(() => undefined));
|
|
const row = await runBoundedWalk(createScheduler());
|
|
const progress = row.progress as HourlyProgressView;
|
|
expect(progress.result).toBe('partial');
|
|
expect(progress.sections).toHaveLength(1);
|
|
expect(progress.sections[0].section).toBe('membership');
|
|
expect(progress.sections[0].errorCode).toBe('TBANK_UNAVAILABLE');
|
|
});
|
|
|
|
it('bounds a membership step promise that never resolves', async () => {
|
|
membership.request.mockImplementation(async () => {
|
|
const row = await child('running', null, 'membership');
|
|
return { id: row.id, done: new Promise(() => undefined) };
|
|
});
|
|
const row = await runBoundedWalk(createScheduler());
|
|
const progress = row.progress as HourlyProgressView;
|
|
expect(progress.result).toBe('partial');
|
|
const membershipSection = progress.sections.find(
|
|
(entry) => entry.section === 'membership',
|
|
);
|
|
expect(membershipSection?.errorCode).toBe('DEADLINE');
|
|
});
|
|
|
|
it('settles the attempt when a post-wait refresh never resolves', async () => {
|
|
events.get.mockImplementation(() => new Promise(() => undefined));
|
|
const row = await runBoundedWalk(createScheduler());
|
|
const progress = row.progress as HourlyProgressView;
|
|
expect(progress.result).toBe('succeeded');
|
|
expect(progress.sections).toEqual([]);
|
|
});
|
|
|
|
it('recovers a lost settle and unblocks the next attempt at its term', async () => {
|
|
const settleSpy = jest
|
|
.spyOn(repository, 'settleHourly')
|
|
.mockRejectedValueOnce(new Error('db'));
|
|
const scheduler = createScheduler();
|
|
await scheduler.tick();
|
|
for (let i = 0; i < 40 && settleSpy.mock.calls.length === 0; i++) {
|
|
await immediate();
|
|
await jest.advanceTimersByTimeAsync(50);
|
|
}
|
|
expect(settleSpy.mock.calls).toHaveLength(1);
|
|
const open = await hourlyRow();
|
|
expect(open?.state).toBe('running');
|
|
await pump(20);
|
|
await scheduler.tick();
|
|
await pump(20);
|
|
const closed = await hourlyRow();
|
|
expect(closed?.state).toBe('complete');
|
|
expect((closed?.progress as HourlyProgressView).result).toBe('partial');
|
|
expect(closed?.errorCode).toBe('INTERRUPTED');
|
|
await jest.advanceTimersByTimeAsync(61 * 60 * 1000);
|
|
await scheduler.tick();
|
|
await pump(30);
|
|
const next = await hourlyRow();
|
|
expect(next?.attemptId).not.toBe(closed?.attemptId);
|
|
expect(next?.state).not.toBe('running');
|
|
expect((next?.progress as HourlyProgressView).result).toBe('succeeded');
|
|
});
|
|
|
|
it('closes an abandoned foreign-process attempt and keeps the normal term', async () => {
|
|
const stuck = await child('complete', null, 'membership');
|
|
const deadAttemptId = randomUUID();
|
|
await prisma.tbankSyncWork.create({
|
|
data: {
|
|
id: randomUUID(),
|
|
userId: USER,
|
|
revision: REVISION,
|
|
scopeKey: 'hourly',
|
|
attemptId: deadAttemptId,
|
|
processRunId: 'dead-process',
|
|
priority: 0,
|
|
intent: {},
|
|
state: 'running',
|
|
progress: {
|
|
children: [{ id: stuck.id, section: 'membership' }],
|
|
attempts: { [stuck.id]: stuck.attemptId },
|
|
sections: [],
|
|
},
|
|
},
|
|
});
|
|
const scheduler = createScheduler();
|
|
await runWalk(scheduler);
|
|
const closed = await hourlyRow();
|
|
expect(closed?.attemptId).toBe(deadAttemptId);
|
|
expect(closed?.state).not.toBe('running');
|
|
expect((closed?.progress as HourlyProgressView).result).toBe('partial');
|
|
expect(closed?.errorCode).toBe('INTERRUPTED');
|
|
const hourlyCount = await prisma.tbankSyncWork.count({
|
|
where: { userId: USER, scopeKey: 'hourly' },
|
|
});
|
|
expect(hourlyCount).toBe(1);
|
|
await jest.advanceTimersByTimeAsync(61 * 60 * 1000);
|
|
await scheduler.tick();
|
|
await pump(30);
|
|
const next = await hourlyRow();
|
|
expect(next?.attemptId).not.toBe(deadAttemptId);
|
|
expect(next?.state).not.toBe('running');
|
|
});
|
|
|
|
describe('hourly term eligibility', () => {
|
|
const at = (h: number, m = 0) => new Date(Date.UTC(2026, 0, 1, h, m));
|
|
const seed = (
|
|
state: 'complete' | 'failed' | 'running',
|
|
start: Date,
|
|
end: Date,
|
|
processRunId = 'other-process',
|
|
) =>
|
|
prisma.tbankSyncWork.create({
|
|
data: {
|
|
id: randomUUID(),
|
|
userId: USER,
|
|
revision: REVISION,
|
|
scopeKey: 'hourly',
|
|
attemptId: randomUUID(),
|
|
processRunId,
|
|
priority: 0,
|
|
intent: {},
|
|
progress: {},
|
|
state,
|
|
createdAt: start,
|
|
updatedAt: end,
|
|
},
|
|
});
|
|
const reserve = async (when: Date) => {
|
|
jest.setSystemTime(when);
|
|
return repository.reserveHourly(USER, 'this-process', 3_600_000);
|
|
};
|
|
|
|
it.each([
|
|
['short attempt', 'complete', at(10), at(10, 5), at(10, 59), at(11)],
|
|
['failed attempt', 'failed', at(10), at(10, 5), at(10, 59), at(11)],
|
|
['walk past one term', 'complete', at(10), at(11, 10), at(11, 59), at(12)],
|
|
['walk past two terms', 'complete', at(10), at(12, 10), at(12, 59), at(13)],
|
|
['walk ending on a term', 'complete', at(10), at(11), at(10, 59), at(11)],
|
|
] as const)('%s', async (_name, state, start, end, early, due) => {
|
|
await seed(state, start, end);
|
|
expect(await reserve(early)).toBeNull();
|
|
const reserved = await reserve(due);
|
|
expect(reserved?.work.state).toBe('running');
|
|
expect(reserved?.work.createdAt).toEqual(due);
|
|
});
|
|
|
|
it('starts a new revision at once', async () => {
|
|
await seed('complete', at(10), at(10, 5));
|
|
await prisma.tbankConnection.deleteMany({ where: { userId: USER } });
|
|
await prisma.tbankConnection.create({
|
|
data: {
|
|
userId: USER,
|
|
revision: 'next-revision',
|
|
encryptedToken: tokens.encrypt('synthetic'),
|
|
connectedAt: at(10, 10),
|
|
},
|
|
});
|
|
expect((await reserve(at(10, 10)))?.work.revision).toBe('next-revision');
|
|
});
|
|
|
|
it('closes a dead process attempt without early retry and runs one catch-up after downtime', async () => {
|
|
await seed('running', at(10), at(10, 1), 'dead-process');
|
|
expect(await reserve(at(10, 30))).toBeNull();
|
|
expect((await hourlyRow())?.errorCode).toBe('INTERRUPTED');
|
|
expect(await reserve(at(10, 59))).toBeNull();
|
|
expect((await reserve(at(14, 30)))?.work.createdAt).toEqual(at(14, 30));
|
|
expect(await reserve(at(14, 31))).toBeNull();
|
|
expect(
|
|
await prisma.tbankSyncWork.count({
|
|
where: { userId: USER, scopeKey: 'hourly' },
|
|
}),
|
|
).toBe(1);
|
|
});
|
|
|
|
it('does not reserve while this process still runs the attempt', async () => {
|
|
await seed('running', at(10), at(10, 1), 'this-process');
|
|
expect(await reserve(at(12, 30))).toBeNull();
|
|
});
|
|
});
|
|
|
|
it('does not overlap a running attempt with a second walk', async () => {
|
|
operations.request.mockImplementation(async () => {
|
|
const row = await child('running', null, 'operations');
|
|
return row.id;
|
|
});
|
|
const scheduler = createScheduler();
|
|
await scheduler.tick();
|
|
await pump();
|
|
expect((await hourlyRow())?.state).toBe('running');
|
|
await scheduler.tick();
|
|
await pump();
|
|
const hourlyCount = await prisma.tbankSyncWork.count({
|
|
where: { userId: USER, scopeKey: 'hourly' },
|
|
});
|
|
expect(hourlyCount).toBe(1);
|
|
expect((await hourlyRow())?.state).toBe('running');
|
|
await jest.advanceTimersByTimeAsync(16 * 60 * 1000);
|
|
await pump(20);
|
|
expect((await hourlyRow())?.state).not.toBe('running');
|
|
});
|
|
|
|
it('does not start a new attempt before the next hourly term', async () => {
|
|
const scheduler = createScheduler();
|
|
const row = await runWalk(scheduler);
|
|
const firstAttemptId = row.attemptId;
|
|
expect(row.state).toBe('complete');
|
|
await scheduler.tick();
|
|
expect((await hourlyRow())?.attemptId).toBe(firstAttemptId);
|
|
await jest.advanceTimersByTimeAsync(59 * 60 * 1000);
|
|
await scheduler.tick();
|
|
expect((await hourlyRow())?.attemptId).toBe(firstAttemptId);
|
|
await jest.advanceTimersByTimeAsync(2 * 60 * 1000);
|
|
await scheduler.tick();
|
|
await pump(30);
|
|
expect((await hourlyRow())?.attemptId).not.toBe(firstAttemptId);
|
|
});
|
|
|
|
it('counts a late step result before the attempt outcome is recorded', async () => {
|
|
const late: { id: string }[] = [];
|
|
operations.request.mockImplementation(async () => {
|
|
const row = await child('running', null, 'operations');
|
|
late.push(row);
|
|
return row.id;
|
|
});
|
|
const scheduler = createScheduler();
|
|
await scheduler.tick();
|
|
await pump();
|
|
assert.ok(late.length > 0);
|
|
for (const row of late) {
|
|
await prisma.tbankSyncWork.update({
|
|
where: { id: row.id },
|
|
data: { state: 'complete' },
|
|
});
|
|
}
|
|
let settled = false;
|
|
for (let i = 0; i < 100 && !settled; i++) {
|
|
await immediate();
|
|
await jest.advanceTimersByTimeAsync(200);
|
|
const current = await hourlyRow();
|
|
settled = Boolean(current && current.state !== 'running');
|
|
}
|
|
const row = await hourlyRow();
|
|
expect(row).not.toBeNull();
|
|
expect(row?.state).not.toBe('running');
|
|
expect((row?.progress as HourlyProgressView).result).toBe('succeeded');
|
|
});
|
|
|
|
it('treats a step superseded by a successor as stale and keeps CAS fencing', async () => {
|
|
const stale = await child('complete', null, 'membership');
|
|
const attemptId = randomUUID();
|
|
const id = randomUUID();
|
|
await prisma.tbankSyncWork.create({
|
|
data: {
|
|
id,
|
|
userId: USER,
|
|
revision: REVISION,
|
|
scopeKey: 'hourly',
|
|
attemptId,
|
|
processRunId: 'dead-process',
|
|
priority: 0,
|
|
intent: {},
|
|
state: 'running',
|
|
progress: {
|
|
children: [{ id: stale.id, section: 'membership' }],
|
|
attempts: { [stale.id]: randomUUID() },
|
|
sections: [],
|
|
},
|
|
},
|
|
});
|
|
const foreign = await repository.settleHourly(
|
|
{ userId: USER, revision: REVISION, id, attemptId: randomUUID() },
|
|
'INTERRUPTED',
|
|
);
|
|
expect(foreign.count).toBe(0);
|
|
expect((await hourlyRow())?.state).toBe('running');
|
|
const closed = await repository.settleHourly(
|
|
{ userId: USER, revision: REVISION, id, attemptId },
|
|
'INTERRUPTED',
|
|
);
|
|
expect(closed.count).toBe(1);
|
|
const progress = (await hourlyRow())?.progress as HourlyProgressView;
|
|
expect(progress.result).toBe('failed');
|
|
expect(progress.sections[0].errorCode).toBeNull();
|
|
const late = await repository.settleHourly(
|
|
{ userId: USER, revision: REVISION, id, attemptId },
|
|
'INTERRUPTED',
|
|
);
|
|
expect(late.count).toBe(0);
|
|
});
|
|
|
|
it('creates no reservations after a controlled shutdown', async () => {
|
|
const scheduler = createScheduler();
|
|
scheduler.onModuleInit();
|
|
scheduler.onModuleDestroy();
|
|
await jest.advanceTimersByTimeAsync(2 * 60 * 60 * 1000);
|
|
await pump(5);
|
|
expect(await prisma.tbankSyncWork.count({ where: { userId: USER } })).toBe(0);
|
|
});
|
|
|
|
it('closes the operations step durably when its finalizer cannot record', async () => {
|
|
await prisma.tbankConnection.update({
|
|
where: { userId: USER },
|
|
data: { membershipAt: new Date() },
|
|
});
|
|
await prisma.tbankAccount.create({
|
|
data: {
|
|
userId: USER,
|
|
revision: REVISION,
|
|
accountId: 'a',
|
|
type: 'BROKER',
|
|
name: 'Test',
|
|
upstreamOrder: 0,
|
|
},
|
|
});
|
|
const accessReal = new TbankAccessService(prisma, tokens);
|
|
const operationsRepository = new BrokerageOperationsRepository(prisma);
|
|
const sync = new BrokerageOperationsSyncService(
|
|
operationsRepository,
|
|
accessReal,
|
|
{ getOperationsByCursor: jest.fn() } as never,
|
|
logger as never,
|
|
);
|
|
jest
|
|
.spyOn(operationsRepository, 'pending')
|
|
.mockRejectedValueOnce(new Error('db'));
|
|
jest
|
|
.spyOn(operationsRepository, 'fail')
|
|
.mockRejectedValueOnce(new Error('db'));
|
|
const id = await sync.request(
|
|
USER,
|
|
[
|
|
{
|
|
accountId: 'a',
|
|
from: new Date(Date.UTC(2026, 0, 1)),
|
|
to: new Date(),
|
|
},
|
|
] as never,
|
|
true,
|
|
'screen',
|
|
REVISION,
|
|
);
|
|
assert.ok(id);
|
|
await pump(40);
|
|
const row = await prisma.tbankSyncWork.findFirst({
|
|
where: { userId: USER, id },
|
|
});
|
|
expect(row?.state).toBe('failed');
|
|
});
|
|
|
|
it('finishes open catalog steps durably when publication fails mid-batch', async () => {
|
|
const accessReal = new TbankAccessService(prisma, tokens);
|
|
const catalogRepository = new CatalogDiscoveryRepository(prisma);
|
|
const enrichment = new CatalogEnrichmentService(
|
|
catalogRepository,
|
|
accessReal,
|
|
{ getLastPrices: jest.fn().mockResolvedValue([]) } as never,
|
|
logger as never,
|
|
{} as never,
|
|
);
|
|
jest
|
|
.spyOn(catalogRepository, 'publishSource')
|
|
.mockRejectedValue(new Error('db'));
|
|
await enrichment.requestMany(USER, REVISION, 'share', ['uid'], false, [
|
|
'last-price',
|
|
]);
|
|
await pump(40);
|
|
const rows = await prisma.tbankSyncWork.findMany({
|
|
where: { userId: USER, scopeKey: { not: 'hourly' } },
|
|
});
|
|
assert.ok(rows.length > 0);
|
|
for (const row of rows) expect(row.state).toBe('failed');
|
|
});
|
|
});
|