ksv741
4796d9d3f0
Backend security / security (push) Successful in 8s
Backend verify / verify (push) Successful in 11m9s
Conventional commits / commits (pull_request) Successful in 28s
Backend security / security (pull_request) Successful in 7s
Backend verify / verify (pull_request) Successful in 10m38s
Resolve A01-A10 and C01-C07 from the backend audit. Preserve auth failures, forced candle reconciliation and forecast freshness. Validate MOEX responses and nested HTTP errors; remove unused legacy paths. Verified: 209 unit tests, 262 SQLite e2e tests on Linux, OpenAPI and durability. Record the remaining macOS native SQLite abort in the audit.
715 lines
26 KiB
TypeScript
715 lines
26 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 { PrismaService } from '../src/prisma/prisma.service';
|
|
import { TbankAccessService } from '../src/integrations/tbank/tbank-access.service';
|
|
import type {
|
|
TbankOperationsByCursorRequest,
|
|
TbankOperationsByCursorResponse,
|
|
} from '../src/integrations/tbank/tbank.client';
|
|
import { TbankTokenService } from '../src/integrations/tbank/tbank-token.service';
|
|
import { BrokerageOperationsRepository } from '../src/modules/investments/operations/infrastructure/brokerage-operations.repository';
|
|
import { BrokerageOperationsSyncService } from '../src/modules/investments/operations/application/brokerage-operations-sync.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('joins overlapping account and instrument refresh into one atomic publication group', async () => {
|
|
const gate = deferred<TbankOperationsByCursorResponse>();
|
|
upstream.getOperationsByCursor
|
|
.mockReturnValueOnce(gate.promise)
|
|
.mockResolvedValue({
|
|
hasNext: false,
|
|
items: [{ ...operation(), instrumentUid: 'uid' }],
|
|
});
|
|
const subset = requiredHandle(
|
|
await service.request(
|
|
'owner',
|
|
[{ ...january, instrumentUid: 'uid' }],
|
|
true,
|
|
),
|
|
);
|
|
await delay(5);
|
|
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: true,
|
|
});
|
|
expect(upstream.getOperationsByCursor).not.toHaveBeenCalled();
|
|
upstream.getOperationsByCursor.mockResolvedValue({
|
|
hasNext: false,
|
|
items: [],
|
|
});
|
|
const refresh = await service.readPlan('owner', [target], {
|
|
movingEnd: 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') {
|
|
expect(await prisma.tbankOperation.count()).toBe(1);
|
|
expect(await prisma.tbankSyncStage.count()).toBe(0);
|
|
} else {
|
|
expect(await prisma.tbankOperation.count()).toBe(0);
|
|
expect(await prisma.tbankSyncStage.count()).toBeGreaterThan(0);
|
|
upstream.getOperationsByCursor.mockResolvedValueOnce(complete);
|
|
await service.request('owner', [january], true);
|
|
await state(handle, 'complete');
|
|
expect(
|
|
upstream.getOperationsByCursor.mock.calls[0][1].cursor,
|
|
).toBeUndefined();
|
|
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;
|
|
nextCursor: string;
|
|
items: never[];
|
|
}>();
|
|
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: string | null): string {
|
|
if (!value) throw new Error('Expected sync handle');
|
|
return value;
|
|
}
|