backend/test/tbank-ledger-local.e2e-spec.ts

import assert from 'node:assert/strict';
import { INestApplication } from '@nestjs/common';
import { JwtService } from '@nestjs/jwt';
import { Test } from '@nestjs/testing';
import { setTimeout as delay } from 'node:timers/promises';
import type { Server } from 'node:http';
import request from 'supertest';
import { AppModule } from '../src/app.module';
import { configureApp } from '../src/main';
import { PrismaService } from '../src/prisma/prisma.service';
import { InstrumentMetadataRepository } from '../src/integrations/tbank/instrument-metadata.repository';
import { TbankClient } from '../src/integrations/tbank/tbank.client';
import { BrokerageOperationsResourceDto } from '../src/modules/investments/operations/presentation/brokerage-operation.dto';
import { BrokerageOperationsRepository } from '../src/modules/investments/operations/infrastructure/brokerage-operations.repository';
import type { IncomeAnalyticsResourceDto } from '../src/modules/investments/income/income-analytics.dto';

const uid = '11111111-1111-4111-8111-111111111111';
const period = 'from=2025-01-01&to=2025-01-31';
describe('persistent ledger HTTP and cursor (SQLite)', () => {
  let app: INestApplication<Server>;
  let prisma: PrismaService;
  const upstream = {
    validateToken: jest.fn(),
    getOpenAccounts: jest.fn(),
    getOperationsByCursor: jest.fn(),
  };
  const auth = (owner = 'owner') =>
    `Bearer ${new JwtService().sign({ sub: owner, type: 'access' }, { secret: process.env.JWT_SECRET })}`;
  const get = (query = '', account = 'all', owner = 'owner') =>
    request(app.getHttpServer())
      .get(
        `/api/v1/brokerage-accounts/${account}/operations?${period}&${query}`,
      )
      .set('Authorization', auth(owner))
      .timeout(2000);
  const put = (owner = 'owner') =>
    request(app.getHttpServer())
      .put('/api/v1/tbank-connection')
      .set('Authorization', auth(owner))
      .send({ token: 'synthetic' });
  beforeAll(async () => {
    const module = await Test.createTestingModule({ imports: [AppModule] })
      .overrideProvider(TbankClient)
      .useValue(upstream)
      .compile();
    app = module.createNestApplication();
    configureApp(app);
    await app.init();
    await app.listen(0, '127.0.0.1');
    prisma = app.get(PrismaService);
    for (const id of ['owner', 'other'])
      await prisma.user.create({
        data: { id, email: `${id}@example.invalid`, passwordHash: 'synthetic' },
      });
  });
  afterAll(async () => app?.close());
  beforeEach(async () => {
    await prisma.tbankConnection.deleteMany();
    jest.clearAllMocks();
    upstream.validateToken.mockResolvedValue(
      ['a', 'b'].map((id) => ({
        id,
        name: id,
        type: 'ACCOUNT_TYPE_TINKOFF',
        openedDate: { seconds: '1577836800' },
      })),
    );
    upstream.getOperationsByCursor
      .mockReset()
      .mockResolvedValue({ hasNext: false, items: [] });
    await put().expect(200);
    const { revision } = await prisma.tbankConnection.findUniqueOrThrow({
      where: { userId: 'owner' },
    });
    for (const accountId of ['a', 'b']) {
      await prisma.tbankOperationCoverage.create({
        data: {
          userId: 'owner',
          revision,
          accountId,
          from: new Date('2020-01-01Z'),
          to: new Date(),
          asOf: new Date(),
          fetchedAt: new Date(),
        },
      });
      for (const id of ['1', '2', '3'])
        await prisma.tbankOperation.create({
          data: {
            userId: 'owner',
            revision,
            accountId,
            brokerOperationId: id,
            occurredAt: new Date('2025-01-02Z'),
            category: id === '3' ? 'trade' : 'income',
            instrumentUid: uid,
            payload: {
              id,
              instrumentUid: uid,
              date: { seconds: '1735776000' },
              type: id === '3' ? 'OPERATION_TYPE_BUY' : 'OPERATION_TYPE_COUPON',
              payment: { currency: 'rub', units: '1', nano: 1 },
            },
          },
        });
    }
  });
  it('reads ordered pages with identical IDs/timestamps across accounts and no upstream RPC, including after reopen', async () => {
    let cursor: string | null = null;
    const ids: string[] = [];
    do {
      const response = await get(
        `limit=2${cursor ? `&cursor=${encodeURIComponent(cursor)}` : ''}`,
      ).expect(200);
      const result = response.body as BrokerageOperationsResourceDto;
      expect(result.meta.availability).toBe('complete');
      assert(result.data);
      ids.push(
        ...result.data.items.map((item) => `${item.account?.id}/${item.id}`),
      );
      expect(result.data.items[0].payment).toEqual({
        currency: 'RUB',
        value: '1.000000001',
      });
      cursor = result.data.nextCursor;
      if (cursor) await prisma.$disconnect();
    } while (cursor);
    expect(ids).toEqual(['a/1', 'a/2', 'a/3', 'b/1', 'b/2', 'b/3']);
    expect(upstream.getOperationsByCursor).not.toHaveBeenCalled();
    expect(upstream.getOpenAccounts).not.toHaveBeenCalled();
  });
  it('uses the persisted subsecond timestamp for filtering, cursor order and income entries', async () => {
    await prisma.tbankOperation.deleteMany();
    const { revision } = await prisma.tbankConnection.findUniqueOrThrow({
      where: { userId: 'owner' },
    });
    const repository = app.get(BrokerageOperationsRepository);
    const interval = {
      accountId: 'a',
      from: new Date('2025-01-02T00:00:00.500Z'),
      to: new Date('2025-01-02T00:00:01.000Z'),
    };
    const reserved = await repository.request(
      'owner',
      revision,
      [interval],
      'subsecond-test',
      true,
      new Set(),
      1,
    );
    assert(reserved);
    const attempt = {
      userId: 'owner',
      revision,
      id: reserved.work.id,
      attemptId: reserved.work.attemptId,
    };
    await expect(
      repository.stagePage(attempt, interval, undefined, {
        hasNext: false,
        items: [{ id: 'invalid', date: {} }],
      }),
    ).rejects.toThrow('Invalid operation timestamp');
    await repository.stagePage(attempt, interval, undefined, {
      hasNext: false,
      items: [400, 600, 800].map((milliseconds) => ({
        id: String(milliseconds),
        date: { seconds: '1735776000', nanos: milliseconds * 1_000_000 },
        type: 'OPERATION_TYPE_COUPON',
        payment: { currency: 'rub', units: '1', nano: 1 },
      })),
    });
    await repository.publish(attempt, [interval], new Date(), true);
    const saved = await prisma.tbankOperation.findMany({
      orderBy: { occurredAt: 'desc' },
    });
    const expected = [
      { id: '800', occurredAt: '2025-01-02T00:00:00.800Z' },
      { id: '600', occurredAt: '2025-01-02T00:00:00.600Z' },
    ];
    expect(
      saved.map((row) => ({
        id: row.brokerOperationId,
        occurredAt: row.occurredAt.toISOString(),
      })),
    ).toEqual(expected);
    const items: { id: string; occurredAt: string }[] = [];
    let cursor: string | null = null;
    do {
      const response = await request(app.getHttpServer())
        .get(
          `/api/v1/brokerage-accounts/a/operations?from=${interval.from.toISOString()}&to=2025-01-02T00:00:00.900Z&limit=1&sync=observe${cursor ? `&cursor=${encodeURIComponent(cursor)}` : ''}`,
        )
        .set('Authorization', auth())
        .expect(200);
      const result = response.body as BrokerageOperationsResourceDto;
      assert(result.data);
      expect(result.data.items).toHaveLength(1);
      items.push(
        ...result.data.items.map(({ id, occurredAt }) => ({ id, occurredAt })),
      );
      cursor = result.data.nextCursor;
    } while (cursor);
    const income = (
      await request(app.getHttpServer())
        .get(
          `/api/v1/brokerage-accounts/a/income-analytics?${period}&sync=observe`,
        )
        .set('Authorization', auth())
        .expect(200)
    ).body as IncomeAnalyticsResourceDto;
    expect(items).toEqual(expected);
    expect(
      income.data?.entries.map(({ id, occurredAt }) => ({ id, occurredAt })),
    ).toEqual(expected);
    expect(income.data?.totalsByCurrency[0].total).toEqual({
      currency: 'RUB',
      value: '2.000000002',
    });
    expect(upstream.getOperationsByCursor).not.toHaveBeenCalled();
  });
  it.each(['default', 'today', 'today timestamp'])(
    'keeps observed current coverage complete through %s period pagination without RPC',
    async (periodKind) => {
      const observedAt = new Date(Date.now() - 1000);
      await prisma.tbankOperationCoverage.updateMany({
        data: {
          to: observedAt,
          asOf: observedAt,
          fetchedAt: observedAt,
        },
      });
      await prisma.tbankOperation.updateMany({
        data: { occurredAt: new Date(observedAt.valueOf() - 1000) },
      });
      const query =
        periodKind === 'default'
          ? ''
          : `&to=${observedAt.toISOString().slice(0, 10)}${periodKind === 'today timestamp' ? 'T23:59:59.999Z' : ''}`;
      const page = (cursor?: string, observe = false) =>
        request(app.getHttpServer())
          .get(
            `/api/v1/brokerage-accounts/all/operations?${observe ? 'sync=observe&' : ''}limit=1${query}${cursor ? `&cursor=${encodeURIComponent(cursor)}` : ''}`,
          )
          .set('Authorization', auth());
      const first = (await page().expect(200))
        .body as BrokerageOperationsResourceDto;
      expect(first.meta.availability).toBe('complete');
      assert(first.data?.nextCursor);
      const second = (await page(first.data.nextCursor).expect(200))
        .body as BrokerageOperationsResourceDto;
      expect(second.meta.availability).toBe('complete');
      expect(second.meta.coverage?.missingRanges).toEqual([]);
      const observed = (await page(first.data.nextCursor, true).expect(200))
        .body as BrokerageOperationsResourceDto;
      expect(observed.meta.availability).toBe('complete');
      expect(upstream.getOperationsByCursor).not.toHaveBeenCalled();
    },
  );
  it('keeps status versions stable between observational reads of a moving period', async () => {
    const observe = () =>
      request(app.getHttpServer())
        .get('/api/v1/brokerage-accounts/all/operations?sync=observe&limit=100')
        .set('Authorization', auth());
    const first = (await observe().expect(200))
      .body as BrokerageOperationsResourceDto;
    await delay(20);
    const second = (await observe().expect(200))
      .body as BrokerageOperationsResourceDto;
    expect(second.meta.coverage?.requested).not.toEqual(
      first.meta.coverage?.requested,
    );
    expect(second.meta.stateVersion).toBe(first.meta.stateVersion);
    await prisma.tbankOperationCoverage.updateMany({
      data: {
        to: new Date(Date.now() - 60_000),
        asOf: new Date(Date.now() - 60_000),
      },
    });
    const changed = (await observe().expect(200))
      .body as BrokerageOperationsResourceDto;
    expect(changed.meta.stateVersion).not.toBe(second.meta.stateVersion);
    expect(upstream.getOperationsByCursor).not.toHaveBeenCalled();
  });
  it('reads shared metadata from SQLite and changes stateVersion without invalidating a row cursor', async () => {
    const before = (await get('limit=1&sync=observe').expect(200))
      .body as BrokerageOperationsResourceDto;
    assert(before.data?.nextCursor);
    const { revision } = await prisma.tbankConnection.findUniqueOrThrow({
      where: { userId: 'owner' },
    });
    const metadata = new InstrumentMetadataRepository(prisma);
    const reservation = await metadata.reserve(
      { userId: 'owner', revision },
      { instrumentUid: uid },
      'metadata-test',
      new Set(),
    );
    assert(reservation);
    await metadata.publish(
      {
        userId: 'owner',
        revision,
        id: reservation.work.id,
        attemptId: reservation.work.attemptId,
      },
      [
        {
          type: 'uid',
          id: uid,
          data: { uid, name: 'Synthetic', brand: { logoName: 'saved.png' } },
        },
      ],
      new Date(),
    );
    await prisma.$disconnect();
    const after = (await get('limit=1&sync=observe').expect(200))
      .body as BrokerageOperationsResourceDto;
    expect(after.data?.items[0].logoName).toBe('saved.png');
    expect(after.meta.stateVersion).not.toBe(before.meta.stateVersion);
    expect(after.meta.dataRevision).toBe(before.meta.dataRevision);
    await get(
      `limit=1&sync=observe&cursor=${encodeURIComponent(before.data.nextCursor)}`,
    ).expect(200);
    expect(upstream.getOperationsByCursor).not.toHaveBeenCalled();
  });
  it('binds a signed cursor to equivalent filters, user, membership, row revision and connection; status alone does not break it', async () => {
    const first = (
      await get('limit=1&categories=trade,income&sync=observe').expect(200)
    ).body as BrokerageOperationsResourceDto;
    assert(first.data?.nextCursor);
    const cursor = encodeURIComponent(first.data.nextCursor);
    const next = `limit=1&categories=income,trade,income&sync=observe&cursor=${cursor}`;
    await prisma.tbankOperationCoverage.updateMany({
      data: { fetchedAt: new Date(0) },
    });
    await get(next).expect(200);
    await get(next.replace('limit=1', 'limit=2')).expect(400);
    await get(`${next}x`).expect(400);
    await put('other').expect(200);
    await get(next, 'all', 'other').expect(400);
    await prisma.tbankAccount.updateMany({
      where: { userId: 'owner', accountId: 'b' },
      data: { operationsRevision: { increment: 1 } },
    });
    await get(next).expect(400);
    const single = (await get('limit=1&sync=observe', 'a').expect(200))
      .body as BrokerageOperationsResourceDto;
    assert(single.data?.nextCursor);
    const singleNext = `limit=1&sync=observe&cursor=${encodeURIComponent(single.data.nextCursor)}`;
    await prisma.tbankConnection.update({
      where: { userId: 'owner' },
      data: { membershipRevision: { increment: 1 } },
    });
    await get(singleNext, 'a').expect(200);
    await put().expect(200);
    await get(singleNext, 'a').expect(400);
  });
  it('does not confuse pagination/categories/instrument subset with interval completeness', async () => {
    await prisma.tbankOperationCoverage.deleteMany();
    const { revision } = await prisma.tbankConnection.findUniqueOrThrow({
      where: { userId: 'owner' },
    });
    await prisma.tbankOperationCoverage.create({
      data: {
        userId: 'owner',
        revision,
        accountId: 'a',
        instrumentPredicate: uid,
        from: new Date('2025-01-01Z'),
        to: new Date('2025-01-04Z'),
        asOf: new Date('2025-01-04Z'),
        fetchedAt: new Date(),
      },
    });
    const partial = (
      await get(
        `instrumentUid=${uid}&categories=income&limit=100&sync=observe`,
        'a',
      ).expect(200)
    ).body as BrokerageOperationsResourceDto;
    expect(partial.meta.availability).toBe('partial');
    expect(partial.data?.hasNext).toBe(false);
    expect(partial.meta.coverage?.missingRanges.length).toBeGreaterThan(0);
    const whole = (await get('sync=observe', 'a').expect(200))
      .body as BrokerageOperationsResourceDto;
    expect(whole.meta.availability).toBe('missing');
    expect(whole.data).toBeNull();
    expect(upstream.getOperationsByCursor).not.toHaveBeenCalled();
  });
  it('keeps an unknown history boundary explicit even when the shown interval is complete', async () => {
    await prisma.tbankAccount.updateMany({ data: { openedOn: null } });
    const result = (await get('sync=observe', 'a').expect(200))
      .body as BrokerageOperationsResourceDto;
    expect(result.meta.availability).toBe('complete');
    expect(result.meta.coverage?.historyBoundary).toBe('unknown');
    expect(result.meta.coverage?.historyState).toBe('partial');
    expect(upstream.getOperationsByCursor).not.toHaveBeenCalled();
  });

  it('returns cold missing while upstream is blocked, then confirms an empty interval; observe never retries failure', async () => {
    await prisma.tbankOperation.deleteMany();
    await prisma.tbankOperationCoverage.deleteMany();
    let unblock: (value: { hasNext: boolean; items: never[] }) => void = () => {
      throw new Error('not initialized');
    };
    upstream.getOperationsByCursor
      .mockReturnValueOnce(
        new Promise((resolve) => {
          unblock = resolve;
        }),
      )
      .mockRejectedValue(new Error('synthetic outage'));
    const cold = (await get('', 'a').expect(200))
      .body as BrokerageOperationsResourceDto;
    expect(cold.data).toBeNull();
    expect(cold.meta.availability).toBe('missing');
    await get('sync=observe', 'a').expect(200);
    expect(upstream.getOperationsByCursor).toHaveBeenCalledTimes(1);
    unblock({ hasNext: false, items: [] });
    for (let index = 0; index < 300; index += 1) {
      if (!(await prisma.tbankSyncWork.count({ where: { state: 'running' } })))
        break;
      await delay(5);
    }
    const count = upstream.getOperationsByCursor.mock.calls.length;
    const result = (await get('sync=observe', 'a').expect(200))
      .body as BrokerageOperationsResourceDto;
    expect(result.data?.items).toEqual([]);
    expect(result.meta.availability).toBe('complete');
    await get('sync=observe', 'a').expect(200);
    expect(upstream.getOperationsByCursor).toHaveBeenCalledTimes(count);
  });
});