backend/test/tbank-operations-sync.e2e-spec.ts

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('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') {
          // 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;
}