768 lines
28 KiB
TypeScript
768 lines
28 KiB
TypeScript
import { ConfigService } from '@nestjs/config';
|
|
import { fork } from 'node:child_process';
|
|
import { once } from 'node:events';
|
|
import { join } from 'node:path';
|
|
import { setTimeout as delay } from 'node:timers/promises';
|
|
|
|
import type {
|
|
TbankOperationsByCursorRequest,
|
|
TbankOperationsByCursorResponse,
|
|
} from '../src/integrations/tbank/tbank.client';
|
|
|
|
import { TbankAccessService } from '../src/integrations/tbank/tbank-access.service';
|
|
import { TbankTokenService } from '../src/integrations/tbank/tbank-token.service';
|
|
import { BrokerageOperationsSyncService } from '../src/modules/investments/operations/application/brokerage-operations-sync.service';
|
|
import { BrokerageOperationsRepository } from '../src/modules/investments/operations/infrastructure/brokerage-operations.repository';
|
|
import { PrismaService } from '../src/prisma/prisma.service';
|
|
|
|
function deferred<T>() {
|
|
let resolve: (value: T) => void = () => {
|
|
throw new Error('Deferred not initialized');
|
|
};
|
|
const promise = new Promise<T>((done) => {
|
|
resolve = done;
|
|
});
|
|
return { promise, resolve };
|
|
}
|
|
|
|
// External RPCs alone are controlled. Work claims, recovery, staging and
|
|
// financial publication use the actual migrated temporary SQLite.
|
|
describe('Request-driven operations synchronization (SQLite)', () => {
|
|
const prisma = new PrismaService(
|
|
new ConfigService({ DATABASE_URL: process.env.DATABASE_URL }),
|
|
);
|
|
const repository = new BrokerageOperationsRepository(prisma);
|
|
const upstream = {
|
|
getOperationsByCursor: jest.fn<
|
|
Promise<TbankOperationsByCursorResponse>,
|
|
[string, TbankOperationsByCursorRequest]
|
|
>(),
|
|
getOpenAccounts: jest.fn(),
|
|
};
|
|
const tokens = new TbankTokenService(
|
|
new ConfigService({
|
|
TOKEN_ENCRYPTION_KEY: process.env.TOKEN_ENCRYPTION_KEY,
|
|
}),
|
|
);
|
|
const access = new TbankAccessService(prisma, tokens);
|
|
const logger = { event: jest.fn() };
|
|
const createService = () =>
|
|
new BrokerageOperationsSyncService(
|
|
repository,
|
|
access,
|
|
upstream as never,
|
|
logger as never,
|
|
);
|
|
let service: BrokerageOperationsSyncService;
|
|
const january = {
|
|
accountId: 'account',
|
|
from: new Date('2026-01-01Z'),
|
|
to: new Date('2026-02-01Z'),
|
|
};
|
|
const february = {
|
|
...january,
|
|
from: january.to,
|
|
to: new Date('2026-03-01Z'),
|
|
};
|
|
const operation = (value = '10', seconds = '1768000000') => ({
|
|
id: 'id',
|
|
date: { seconds },
|
|
payment: { units: value, nano: 1, currency: 'rub' },
|
|
});
|
|
const complete = { hasNext: false, items: [operation()] };
|
|
async function state(id: string, expected: string) {
|
|
for (let index = 0; index < 500; index += 1) {
|
|
const work = await prisma.tbankSyncWork.findUniqueOrThrow({
|
|
where: { id },
|
|
});
|
|
if (work.state === expected) return work;
|
|
await delay(2);
|
|
}
|
|
throw new Error(`Sync did not reach ${expected}`);
|
|
}
|
|
beforeAll(async () => {
|
|
await prisma.$connect();
|
|
await prisma.user.create({
|
|
data: {
|
|
id: 'owner',
|
|
email: 'owner@example.invalid',
|
|
passwordHash: 'synthetic',
|
|
},
|
|
});
|
|
});
|
|
beforeEach(async () => {
|
|
await prisma.tbankConnection.deleteMany();
|
|
jest.clearAllMocks();
|
|
upstream.getOperationsByCursor.mockReset();
|
|
await prisma.tbankConnection.create({
|
|
data: {
|
|
userId: 'owner',
|
|
revision: 'revision',
|
|
encryptedToken: tokens.encrypt('synthetic'),
|
|
connectedAt: new Date(),
|
|
membershipAt: new Date(),
|
|
accounts: {
|
|
create: {
|
|
accountId: 'account',
|
|
upstreamOrder: 0,
|
|
name: 'Synthetic',
|
|
type: 'BROKERAGE',
|
|
},
|
|
},
|
|
},
|
|
});
|
|
service = createService();
|
|
});
|
|
afterAll(async () => prisma.$disconnect());
|
|
|
|
it.each([[false, 1], [true, 1], [false, 2], [true, 2]] as const)(
|
|
'extends legacy intents in stable priority order (atomic=%s, previousPriority=%s)',
|
|
async (atomic, previousPriority) => {
|
|
const run = 'legacy-run';
|
|
const initial = await repository.request('owner', 'revision', [january, february], run, false, new Set(), previousPriority);
|
|
if (!initial) throw new Error('Missing work');
|
|
const key = (range: typeof january) => JSON.stringify([
|
|
range.accountId, range.from.toISOString(), range.to.toISOString(), '',
|
|
]);
|
|
const march = { ...january, from: february.to, to: new Date('2026-04-01Z') };
|
|
await prisma.tbankSyncWork.update({
|
|
where: { id: initial.work.id },
|
|
data: { intent: { atomic, intervals: [key(february), key(january)] } },
|
|
});
|
|
const extended = await repository.request(
|
|
'owner', 'revision', [january, march], run, false,
|
|
new Set([initial.work.attemptId]), 1,
|
|
);
|
|
expect(extended?.launch).toBe(false);
|
|
expect(extended?.work.id).toBe(initial.work.id);
|
|
expect(extended?.work.priority).toBe(1);
|
|
expect(extended?.work.intent).toEqual({
|
|
atomic,
|
|
atomicIntervals: atomic ? [key(february), key(january), key(march)] : [],
|
|
screenIntervals: previousPriority === 1
|
|
? [key(february), key(january), key(march)]
|
|
: [key(january), key(march)],
|
|
intervals: previousPriority === 1
|
|
? [key(february), key(january), key(march)]
|
|
: [key(january), key(march), key(february)],
|
|
});
|
|
},
|
|
);
|
|
|
|
it('joins overlapping account and instrument refresh into one atomic publication group', async () => {
|
|
const gate = deferred<TbankOperationsByCursorResponse>();
|
|
const started = deferred<void>();
|
|
upstream.getOperationsByCursor
|
|
.mockImplementationOnce(() => {
|
|
started.resolve();
|
|
return gate.promise;
|
|
})
|
|
.mockResolvedValue({
|
|
hasNext: false,
|
|
items: [{ ...operation(), instrumentUid: 'uid' }],
|
|
});
|
|
const subset = requiredHandle(
|
|
await service.request(
|
|
'owner',
|
|
[{ ...january, instrumentUid: 'uid' }],
|
|
true,
|
|
),
|
|
);
|
|
await started.promise;
|
|
const whole = requiredHandle(
|
|
await service.request('owner', [january], true),
|
|
);
|
|
expect(whole).toBe(subset);
|
|
expect(await prisma.tbankOperation.count()).toBe(0);
|
|
gate.resolve({
|
|
hasNext: false,
|
|
items: [{ ...operation(), instrumentUid: 'uid' }],
|
|
});
|
|
await state(whole, 'complete');
|
|
expect(await prisma.tbankOperation.count()).toBe(1);
|
|
expect(await prisma.tbankOperationCoverage.count()).toBe(2);
|
|
expect(upstream.getOperationsByCursor).toHaveBeenCalledTimes(2);
|
|
});
|
|
|
|
it('publishes the forced visible period before unrelated pending history and retains it if history fails', async () => {
|
|
const gate = deferred<TbankOperationsByCursorResponse>();
|
|
const march = {
|
|
...january,
|
|
from: new Date('2026-03-01Z'),
|
|
to: new Date('2026-04-01Z'),
|
|
};
|
|
upstream.getOperationsByCursor
|
|
.mockReturnValueOnce(gate.promise)
|
|
.mockImplementation((_token, request) => {
|
|
if (Number(request.from?.seconds) + 1 === march.from.valueOf() / 1000) {
|
|
return Promise.resolve({
|
|
hasNext: false,
|
|
items: [operation('20', '1773000000')],
|
|
});
|
|
}
|
|
return Promise.reject(new Error('unrelated history unavailable'));
|
|
});
|
|
const history = requiredHandle(
|
|
await service.request('owner', [february, march], false, 'history'),
|
|
);
|
|
await delay(5);
|
|
const screen = requiredHandle(
|
|
await service.request('owner', [march], true),
|
|
);
|
|
expect(screen).toBe(history);
|
|
gate.resolve({ hasNext: false, items: [] });
|
|
await state(screen, 'failed');
|
|
expect(
|
|
(await prisma.tbankOperation.findMany()).map((row) => row.payload),
|
|
).toEqual([operation('20', '1773000000')]);
|
|
const coverage = await repository.coverage('owner', 'revision', [
|
|
'account',
|
|
]);
|
|
expect(coverage.map((range) => [range.from, range.to])).toEqual([
|
|
[march.from, march.to],
|
|
]);
|
|
});
|
|
|
|
it('moves a newly visible interval ahead of pending backfill and records an unknown requested floor', async () => {
|
|
const gate = deferred<TbankOperationsByCursorResponse>();
|
|
upstream.getOperationsByCursor
|
|
.mockReturnValueOnce(gate.promise)
|
|
.mockResolvedValue({ hasNext: false, items: [] });
|
|
const march = {
|
|
...january,
|
|
from: new Date('2026-03-01Z'),
|
|
to: new Date('2026-04-01Z'),
|
|
};
|
|
const history = requiredHandle(
|
|
await service.request(
|
|
'owner',
|
|
[january, february, march],
|
|
false,
|
|
'history',
|
|
),
|
|
);
|
|
await delay(5);
|
|
await service.request('owner', [march], false);
|
|
gate.resolve({ hasNext: false, items: [] });
|
|
await state(history, 'complete');
|
|
expect(
|
|
upstream.getOperationsByCursor.mock.calls.map(
|
|
([, call]) => Number(call.from?.seconds) + 1,
|
|
),
|
|
).toEqual([
|
|
january.from.valueOf() / 1000,
|
|
march.from.valueOf() / 1000,
|
|
february.from.valueOf() / 1000,
|
|
]);
|
|
const account = await prisma.tbankAccount.findFirstOrThrow();
|
|
expect(account.historyBoundary).toBe('unknown');
|
|
expect(account.historyFrom).toEqual(new Date(0));
|
|
});
|
|
|
|
it('never publishes coverage beyond the fixed request time', async () => {
|
|
const before = new Date();
|
|
const from = new Date(
|
|
Date.UTC(before.getUTCFullYear(), before.getUTCMonth(), 1),
|
|
);
|
|
const future = new Date(
|
|
Date.UTC(before.getUTCFullYear(), before.getUTCMonth() + 1, 1),
|
|
);
|
|
upstream.getOperationsByCursor.mockResolvedValue({
|
|
hasNext: false,
|
|
items: [],
|
|
});
|
|
const handle = requiredHandle(
|
|
await service.request(
|
|
'owner',
|
|
[{ accountId: 'account', from, to: future }],
|
|
false,
|
|
),
|
|
);
|
|
const accepted = new Date();
|
|
await state(handle, 'complete');
|
|
const saved = await prisma.tbankOperationCoverage.findFirstOrThrow();
|
|
expect(saved.to.valueOf()).toBeGreaterThanOrEqual(before.valueOf());
|
|
expect(saved.to.valueOf()).toBeLessThanOrEqual(accepted.valueOf());
|
|
expect(saved.asOf).toEqual(saved.to);
|
|
});
|
|
|
|
it('uses persistent coverage without RPC, observes stale data without starting work, and refreshes on request', async () => {
|
|
const now = new Date();
|
|
const from = new Date(Date.UTC(now.getUTCFullYear(), now.getUTCMonth(), 1));
|
|
const asOf = new Date(
|
|
now.valueOf() - Math.min(1000, (now.valueOf() - from.valueOf()) / 2),
|
|
);
|
|
await prisma.tbankOperationCoverage.create({
|
|
data: {
|
|
userId: 'owner',
|
|
revision: 'revision',
|
|
accountId: 'account',
|
|
from,
|
|
to: asOf,
|
|
asOf,
|
|
fetchedAt: now,
|
|
},
|
|
});
|
|
const target = { accountId: 'account', from, to: now };
|
|
const warm = await service.readPlan('owner', [target], { movingEnd: true });
|
|
expect(warm.handle).toBeNull();
|
|
expect(warm.plan?.intervals[0]).toMatchObject({
|
|
availability: 'complete',
|
|
asOf,
|
|
stale: false,
|
|
});
|
|
expect(upstream.getOperationsByCursor).not.toHaveBeenCalled();
|
|
await prisma.tbankOperationCoverage.updateMany({
|
|
data: { fetchedAt: new Date(now.valueOf() - 300_000) },
|
|
});
|
|
const observed = await service.readPlan('owner', [target], {
|
|
movingEnd: true,
|
|
observe: true,
|
|
});
|
|
expect(observed.handle).toBeNull();
|
|
expect(observed.plan?.intervals[0]).toMatchObject({
|
|
availability: 'complete',
|
|
asOf,
|
|
stale: false,
|
|
});
|
|
expect(upstream.getOperationsByCursor).not.toHaveBeenCalled();
|
|
upstream.getOperationsByCursor.mockResolvedValue({
|
|
hasNext: false,
|
|
items: [],
|
|
});
|
|
const refresh = await service.readPlan('owner', [target], {
|
|
movingEnd: true,
|
|
force: true,
|
|
});
|
|
await state(requiredHandle(refresh.handle), 'complete');
|
|
expect(upstream.getOperationsByCursor).toHaveBeenCalledTimes(1);
|
|
expect(
|
|
(
|
|
await service.readPlan('owner', [{ ...target, to: new Date() }], {
|
|
movingEnd: true,
|
|
})
|
|
).handle,
|
|
).toBeNull();
|
|
expect(upstream.getOperationsByCursor).toHaveBeenCalledTimes(1);
|
|
const forced = await service.readPlan(
|
|
'owner',
|
|
[{ ...target, to: new Date() }],
|
|
{ movingEnd: true, force: true },
|
|
);
|
|
await state(requiredHandle(forced.handle), 'complete');
|
|
expect(upstream.getOperationsByCursor).toHaveBeenCalledTimes(2);
|
|
});
|
|
|
|
it('preserves the entire complete quarter across accounts when the second month fails', async () => {
|
|
await prisma.tbankAccount.create({
|
|
data: {
|
|
userId: 'owner',
|
|
revision: 'revision',
|
|
accountId: 'second',
|
|
name: 'Second',
|
|
type: 'BROKERAGE',
|
|
upstreamOrder: 1,
|
|
},
|
|
});
|
|
const quarter = [
|
|
{ ...january, to: new Date('2026-04-01Z') },
|
|
{ ...january, accountId: 'second', to: new Date('2026-04-01Z') },
|
|
];
|
|
let amount = '10';
|
|
let failFebruary = false;
|
|
upstream.getOperationsByCursor.mockImplementation((_token, request) => {
|
|
if (
|
|
failFebruary &&
|
|
request.from?.seconds === String(february.from.valueOf() / 1000 - 1)
|
|
) {
|
|
return Promise.reject(new Error('Second month unavailable'));
|
|
}
|
|
return Promise.resolve({
|
|
hasNext: false,
|
|
items: [
|
|
{
|
|
...operation(amount, String(Number(request.from?.seconds) + 2)),
|
|
id: `payment-${request.from?.seconds}`,
|
|
},
|
|
],
|
|
});
|
|
});
|
|
const handle = requiredHandle(
|
|
await service.request('owner', quarter, false),
|
|
);
|
|
await state(handle, 'complete');
|
|
const before = await prisma.tbankOperation.findMany({
|
|
orderBy: [{ accountId: 'asc' }, { brokerOperationId: 'asc' }],
|
|
});
|
|
const coverage = await prisma.tbankOperationCoverage.findMany();
|
|
expect(before).toHaveLength(6);
|
|
amount = '-20';
|
|
failFebruary = true;
|
|
await service.request('owner', quarter, false);
|
|
await state(handle, 'failed');
|
|
expect(
|
|
await prisma.tbankOperation.findMany({
|
|
orderBy: [{ accountId: 'asc' }, { brokerOperationId: 'asc' }],
|
|
}),
|
|
).toEqual(before);
|
|
expect(await prisma.tbankOperationCoverage.findMany()).toEqual(coverage);
|
|
failFebruary = false;
|
|
await service.request('owner', quarter, false);
|
|
await state(handle, 'complete');
|
|
const after = await prisma.tbankOperation.findMany();
|
|
expect(after).toHaveLength(6);
|
|
expect(
|
|
after.every(
|
|
(row) =>
|
|
(row.payload as { payment: { units: string } }).payment.units ===
|
|
'-20',
|
|
),
|
|
).toBe(true);
|
|
expect(
|
|
(await prisma.tbankAccount.findMany()).map(
|
|
(account) => account.operationsRevision,
|
|
),
|
|
).toEqual([4, 4]);
|
|
});
|
|
|
|
it.each(['between-pages', 'before-publish', 'after-publish'])(
|
|
'survives SIGKILL %s without startup recovery or lost publication',
|
|
async (checkpoint) => {
|
|
const child = fork(
|
|
join(__dirname, 'fixtures/tbank-sync-crash.cjs'),
|
|
[checkpoint],
|
|
{ silent: true },
|
|
);
|
|
let stderr = '';
|
|
child.stderr?.on('data', (chunk: Buffer) => {
|
|
stderr += chunk.toString();
|
|
});
|
|
const exited = once(child, 'exit');
|
|
try {
|
|
const handle = await new Promise<string>((resolve, reject) => {
|
|
const timer = setTimeout(
|
|
() => reject(new Error(`Crash checkpoint timeout: ${stderr}`)),
|
|
5000,
|
|
);
|
|
child.once('exit', () => {
|
|
clearTimeout(timer);
|
|
reject(new Error(`Worker exited early: ${stderr}`));
|
|
});
|
|
child.on('message', (message: unknown) => {
|
|
if (
|
|
message &&
|
|
typeof message === 'object' &&
|
|
'handle' in message &&
|
|
typeof message.handle === 'string'
|
|
) {
|
|
clearTimeout(timer);
|
|
resolve(message.handle);
|
|
}
|
|
if (message && typeof message === 'object' && 'error' in message) {
|
|
clearTimeout(timer);
|
|
reject(new Error(String(message.error)));
|
|
}
|
|
});
|
|
});
|
|
child.kill('SIGKILL');
|
|
await exited;
|
|
service = createService();
|
|
const saved = await service.observe('owner', handle);
|
|
expect(saved.state).toBe(
|
|
checkpoint === 'after-publish' ? 'complete' : 'running',
|
|
);
|
|
await delay(10);
|
|
expect(upstream.getOperationsByCursor).not.toHaveBeenCalled();
|
|
if (checkpoint === 'after-publish') {
|
|
// eslint-disable-next-line jest/no-conditional-expect -- The parameterized fixture chooses the scenario; each branch asserts its outcome and the test also asserts common invariants.
|
|
expect(await prisma.tbankOperation.count()).toBe(1);
|
|
// eslint-disable-next-line jest/no-conditional-expect -- The parameterized fixture chooses the scenario; each branch asserts its outcome and the test also asserts common invariants.
|
|
expect(await prisma.tbankSyncStage.count()).toBe(0);
|
|
} else {
|
|
// eslint-disable-next-line jest/no-conditional-expect -- The parameterized fixture chooses the scenario; each branch asserts its outcome and the test also asserts common invariants.
|
|
expect(await prisma.tbankOperation.count()).toBe(0);
|
|
// eslint-disable-next-line jest/no-conditional-expect -- The parameterized fixture chooses the scenario; each branch asserts its outcome and the test also asserts common invariants.
|
|
expect(await prisma.tbankSyncStage.count()).toBeGreaterThan(0);
|
|
upstream.getOperationsByCursor.mockResolvedValueOnce(complete);
|
|
await service.request('owner', [january], true);
|
|
await state(handle, 'complete');
|
|
// eslint-disable-next-line jest/no-conditional-expect -- The parameterized fixture chooses the scenario; each branch asserts its outcome and the test also asserts common invariants.
|
|
expect(
|
|
upstream.getOperationsByCursor.mock.calls[0][1].cursor,
|
|
).toBeUndefined();
|
|
// eslint-disable-next-line jest/no-conditional-expect -- The parameterized fixture chooses the scenario; each branch asserts its outcome and the test also asserts common invariants.
|
|
expect(await prisma.tbankOperation.count()).toBe(1);
|
|
}
|
|
} finally {
|
|
if (child.exitCode === null && child.signalCode === null) {
|
|
child.kill('SIGKILL');
|
|
}
|
|
await exited;
|
|
}
|
|
},
|
|
);
|
|
|
|
it('joins a request bridging active groups and keeps both original handles observable', async () => {
|
|
const blocked = deferred<typeof complete>();
|
|
upstream.getOperationsByCursor
|
|
.mockReturnValueOnce(blocked.promise)
|
|
.mockResolvedValue({ hasNext: false, items: [] });
|
|
const early = { ...january, to: new Date('2026-01-10Z') };
|
|
const late = { ...january, from: new Date('2026-01-20Z') };
|
|
const first = requiredHandle(await service.request('owner', [early], true));
|
|
await delay(5);
|
|
const second = requiredHandle(await service.request('owner', [late], true));
|
|
const merged = requiredHandle(
|
|
await service.request('owner', [january], true),
|
|
);
|
|
expect([first, second]).toContain(merged);
|
|
blocked.resolve({ hasNext: false, items: [] });
|
|
await state(merged, 'complete');
|
|
expect((await service.observe('owner', first)).state).toBe('complete');
|
|
expect((await service.observe('owner', second)).state).toBe('complete');
|
|
expect(await prisma.tbankOperationCoverage.count()).toBe(1);
|
|
expect(upstream.getOperationsByCursor).toHaveBeenCalledTimes(2);
|
|
});
|
|
|
|
it('extends an overlapping atomic group without rereading its intersection or publishing a partial result', async () => {
|
|
const first = deferred<typeof complete>();
|
|
const remainder = deferred<typeof complete>();
|
|
const early = { ...january, to: new Date('2026-01-20Z') };
|
|
const late = { ...january, from: new Date('2026-01-10Z') };
|
|
upstream.getOperationsByCursor
|
|
.mockReturnValueOnce(first.promise)
|
|
.mockReturnValueOnce(remainder.promise);
|
|
const handle = requiredHandle(
|
|
await service.request('owner', [early], true),
|
|
);
|
|
await delay(5);
|
|
expect(await service.request('owner', [late], true)).toBe(handle);
|
|
first.resolve({ hasNext: false, items: [] });
|
|
for (
|
|
let index = 0;
|
|
index < 500 && upstream.getOperationsByCursor.mock.calls.length < 2;
|
|
index += 1
|
|
) {
|
|
await delay(2);
|
|
}
|
|
expect(upstream.getOperationsByCursor).toHaveBeenCalledTimes(2);
|
|
expect(upstream.getOperationsByCursor.mock.calls[1][1].from).toEqual({
|
|
seconds: String(early.to.valueOf() / 1000 - 1),
|
|
});
|
|
expect(await prisma.tbankOperationCoverage.count()).toBe(0);
|
|
remainder.resolve({ hasNext: false, items: [] });
|
|
await state(handle, 'complete');
|
|
expect(await prisma.tbankOperationCoverage.count()).toBe(2);
|
|
});
|
|
|
|
it('lets new screen work overtake history between provider pages', async () => {
|
|
const first = deferred<{
|
|
hasNext: boolean;
|
|
items: never[];
|
|
nextCursor: string;
|
|
}>();
|
|
upstream.getOperationsByCursor
|
|
.mockReturnValueOnce(first.promise)
|
|
.mockResolvedValue({ hasNext: false, items: [] });
|
|
const history = requiredHandle(
|
|
await service.request('owner', [january], false, 'history'),
|
|
);
|
|
await delay(5);
|
|
const screen = requiredHandle(
|
|
await service.request('owner', [february], true),
|
|
);
|
|
first.resolve({ hasNext: true, nextCursor: 'next', items: [] });
|
|
await state(screen, 'complete');
|
|
await state(history, 'complete');
|
|
expect(upstream.getOperationsByCursor).toHaveBeenCalledTimes(3);
|
|
expect(upstream.getOperationsByCursor.mock.calls[1][1].from).toEqual({
|
|
seconds: String(february.from.valueOf() / 1000 - 1),
|
|
});
|
|
expect(upstream.getOperationsByCursor.mock.calls[2][1].cursor).toBe('next');
|
|
});
|
|
|
|
it('keeps published initial windows on recovery and restarts only the unfinished month', async () => {
|
|
const blocked = deferred<typeof complete>();
|
|
upstream.getOperationsByCursor
|
|
.mockResolvedValueOnce(complete)
|
|
.mockReturnValueOnce(blocked.promise);
|
|
const handle = requiredHandle(
|
|
await service.request('owner', [january, february], false),
|
|
);
|
|
for (
|
|
let index = 0;
|
|
index < 500 && upstream.getOperationsByCursor.mock.calls.length < 2;
|
|
index += 1
|
|
) {
|
|
await delay(2);
|
|
}
|
|
expect(await prisma.tbankOperation.count()).toBe(1);
|
|
const before = await prisma.tbankOperationCoverage.findMany();
|
|
expect(before).toHaveLength(1);
|
|
service = createService();
|
|
upstream.getOperationsByCursor.mockResolvedValueOnce({
|
|
hasNext: false,
|
|
items: [],
|
|
});
|
|
await service.request('owner', [january, february], false);
|
|
await state(handle, 'complete');
|
|
expect(upstream.getOperationsByCursor.mock.calls[2][1]).toMatchObject({
|
|
from: { seconds: String(february.from.valueOf() / 1000 - 1) },
|
|
});
|
|
expect(
|
|
(
|
|
await prisma.tbankOperationCoverage.findFirstOrThrow({
|
|
where: { from: january.from },
|
|
})
|
|
).fetchedAt,
|
|
).toEqual(before[0].fetchedAt);
|
|
blocked.resolve(complete);
|
|
await delay(10);
|
|
expect(await prisma.tbankOperation.count()).toBe(1);
|
|
expect(await prisma.tbankOperationCoverage.count()).toBe(2);
|
|
});
|
|
|
|
it('persists current invalid credentials and rejects work for an inaccessible account before RPC', async () => {
|
|
await expect(
|
|
service.request('owner', [{ ...january, accountId: 'other' }], true),
|
|
).rejects.toMatchObject({ status: 404 });
|
|
expect(upstream.getOperationsByCursor).not.toHaveBeenCalled();
|
|
upstream.getOperationsByCursor.mockRejectedValueOnce(
|
|
new Error('INVALID_TBANK_TOKEN'),
|
|
);
|
|
const handle = requiredHandle(
|
|
await service.request('owner', [january], true),
|
|
);
|
|
expect((await state(handle, 'failed')).errorCode).toBe(
|
|
'INVALID_TBANK_TOKEN',
|
|
);
|
|
expect(
|
|
(
|
|
await prisma.tbankConnection.findUniqueOrThrow({
|
|
where: { userId: 'owner' },
|
|
})
|
|
).invalidToken,
|
|
).toBe(true);
|
|
await expect(
|
|
service.request('owner', [january], true),
|
|
).rejects.toMatchObject({ status: 400 });
|
|
expect(upstream.getOperationsByCursor).toHaveBeenCalledTimes(1);
|
|
});
|
|
|
|
it('coalesces twenty requests, observes without RPC and completes only after the last page', async () => {
|
|
const blocked = deferred<typeof complete>();
|
|
upstream.getOperationsByCursor.mockReturnValueOnce(blocked.promise);
|
|
const handles = await Promise.all(
|
|
Array.from({ length: 20 }, () =>
|
|
service.request('owner', [january], true),
|
|
),
|
|
);
|
|
expect(new Set(handles).size).toBe(1);
|
|
const handle = requiredHandle(handles[0]);
|
|
await delay(5);
|
|
expect(upstream.getOperationsByCursor).toHaveBeenCalledTimes(1);
|
|
expect((await service.observe('owner', handle)).state).toBe('running');
|
|
expect(await prisma.tbankOperation.count()).toBe(0);
|
|
blocked.resolve(complete);
|
|
await state(handle, 'complete');
|
|
expect(await prisma.tbankOperation.count()).toBe(1);
|
|
await Promise.all(
|
|
Array.from({ length: 20 }, () => service.observe('owner', handle)),
|
|
);
|
|
expect(upstream.getOperationsByCursor).toHaveBeenCalledTimes(1);
|
|
expect(upstream.getOpenAccounts).not.toHaveBeenCalled();
|
|
expect(upstream.getOperationsByCursor.mock.calls[0][1]).toMatchObject({
|
|
from: { seconds: String(january.from.valueOf() / 1000 - 1) },
|
|
to: { seconds: String(january.to.valueOf() / 1000 + 1) },
|
|
state: 'OPERATION_STATE_EXECUTED',
|
|
limit: 1000,
|
|
});
|
|
});
|
|
|
|
it('keeps the full old result on a second-month failure, has no retry loop and resumes only on request', async () => {
|
|
upstream.getOperationsByCursor
|
|
.mockResolvedValueOnce(complete)
|
|
.mockResolvedValueOnce({ hasNext: false, items: [] });
|
|
const handle = requiredHandle(
|
|
await service.request('owner', [january, february], false),
|
|
);
|
|
await state(handle, 'complete');
|
|
const oldCoverage = await prisma.tbankOperationCoverage.findMany();
|
|
upstream.getOperationsByCursor
|
|
.mockResolvedValueOnce({ hasNext: false, items: [operation('-20')] })
|
|
.mockRejectedValueOnce(new Error('synthetic outage'));
|
|
await service.request('owner', [january, february], false);
|
|
await state(handle, 'failed');
|
|
expect((await prisma.tbankOperation.findFirstOrThrow()).payload).toEqual(
|
|
operation(),
|
|
);
|
|
expect(await prisma.tbankOperationCoverage.findMany()).toEqual(oldCoverage);
|
|
const calls = upstream.getOperationsByCursor.mock.calls.length;
|
|
await service.observe('owner', handle);
|
|
await delay(10);
|
|
expect(upstream.getOperationsByCursor).toHaveBeenCalledTimes(calls);
|
|
upstream.getOperationsByCursor
|
|
.mockResolvedValueOnce({ hasNext: false, items: [operation('-20')] })
|
|
.mockResolvedValueOnce({ hasNext: false, items: [] });
|
|
await service.request('owner', [january, february], false);
|
|
await state(handle, 'complete');
|
|
expect((await prisma.tbankOperation.findFirstOrThrow()).payload).toEqual(
|
|
operation('-20'),
|
|
);
|
|
});
|
|
|
|
it('starts an unfinished interval from page one after a new run request and ignores late old pages', async () => {
|
|
const blocked = deferred<typeof complete>();
|
|
upstream.getOperationsByCursor
|
|
.mockResolvedValueOnce({
|
|
hasNext: true,
|
|
nextCursor: 'provider-cursor',
|
|
items: [operation()],
|
|
})
|
|
.mockReturnValueOnce(blocked.promise);
|
|
const handle = requiredHandle(
|
|
await service.request('owner', [january], true),
|
|
);
|
|
for (
|
|
let index = 0;
|
|
index < 500 && upstream.getOperationsByCursor.mock.calls.length < 2;
|
|
index += 1
|
|
) {
|
|
await delay(2);
|
|
}
|
|
expect(upstream.getOperationsByCursor).toHaveBeenCalledTimes(2);
|
|
const before = await prisma.tbankSyncWork.findUniqueOrThrow({
|
|
where: { id: handle },
|
|
});
|
|
service = createService();
|
|
await service.observe('owner', handle);
|
|
await delay(5);
|
|
expect(upstream.getOperationsByCursor).toHaveBeenCalledTimes(2);
|
|
upstream.getOperationsByCursor.mockResolvedValueOnce({
|
|
hasNext: false,
|
|
items: [operation('30')],
|
|
});
|
|
await service.request('owner', [january], true);
|
|
await state(handle, 'complete');
|
|
const after = await prisma.tbankSyncWork.findUniqueOrThrow({
|
|
where: { id: handle },
|
|
});
|
|
expect(after.attemptId).not.toBe(before.attemptId);
|
|
expect(after.processRunId).not.toBe(before.processRunId);
|
|
expect(
|
|
upstream.getOperationsByCursor.mock.calls[2][1].cursor,
|
|
).toBeUndefined();
|
|
blocked.resolve(complete);
|
|
await delay(10);
|
|
expect((await prisma.tbankOperation.findFirstOrThrow()).payload).toEqual(
|
|
operation('30'),
|
|
);
|
|
expect(await prisma.tbankSyncStage.count()).toBe(0);
|
|
});
|
|
});
|
|
|
|
function requiredHandle(value: null | string): string {
|
|
if (!value) throw new Error('Expected sync handle');
|
|
return value;
|
|
}
|