ksv741
d6e6c4ace5
Backend security / security (push) Successful in 8s
Backend verify / verify (push) Failing after 10m45s
Conventional commits / commits (pull_request) Successful in 29s
Backend security / security (pull_request) Successful in 8s
Backend verify / verify (pull_request) Failing after 10m37s
Store account membership, portfolio, operations, forecasts and market sources in SQLite. Coalesce scoped refreshes, fence connection revisions, preserve complete financial snapshots and stabilize operation period state. BREAKING CHANGE: investment reads return data/meta resource envelopes and locally persisted snapshots; consumers must use the matching OpenAPI contract and explicit synchronization endpoints.
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, upstream as never);
|
|
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;
|
|
}
|