backend/test/tbank-catalog-discovery.e2e-spec.ts

import assert from 'node:assert/strict';
import { ConfigService } from '@nestjs/config';
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 { TbankTokenService } from '../src/integrations/tbank/tbank-token.service';
import { CatalogDiscoveryService } from '../src/modules/investments/instruments/application/catalog-discovery.service';
import {
  CatalogDiscoveryRepository,
  type CatalogInstrument,
  type CatalogQuery,
} from '../src/modules/investments/instruments/infrastructure/catalog-discovery.repository';

// Real transactions prove query coverage, private work ownership and late-publication fences.
describe('shared catalog discovery (SQLite)', () => {
  const prisma = new PrismaService(
    new ConfigService({ DATABASE_URL: process.env.DATABASE_URL }),
  );
  const tokens = new TbankTokenService(
    new ConfigService({
      TOKEN_ENCRYPTION_KEY: process.env.TOKEN_ENCRYPTION_KEY,
    }) as never,
  );
  const upstream = { findShares: jest.fn(), findBonds: jest.fn() };
  const access = new TbankAccessService(prisma, tokens, upstream as never);
  const repository = new CatalogDiscoveryRepository(prisma);
  const createService = () =>
    new CatalogDiscoveryService(
      repository,
      access,
      upstream as never,
      { event: jest.fn() } as never,
      { requestMany: jest.fn().mockResolvedValue([]) } as never,
    );
  let service: CatalogDiscoveryService;
  const query: CatalogQuery = { instrumentType: 'share', query: 'SBER' };
  const instrument = {
    uid: 'uid',
    name: 'Synthetic share',
    ticker: 'SBER',
    isin: 'TEST',
    instrumentKind: 'INSTRUMENT_TYPE_SHARE',
  };
  const seed = (userId: string) =>
    prisma.tbankConnection.create({
      data: {
        userId,
        revision: `${userId}-revision`,
        encryptedToken: tokens.encrypt(`synthetic-${userId}`),
        connectedAt: new Date(),
      },
    });
  async function settled() {
    for (let i = 0; i < 300; i++) {
      if (!(await prisma.tbankSyncWork.count({ where: { state: 'running' } })))
        return;
      await delay(5);
    }
    throw new Error('Catalog work did not settle');
  }
  beforeAll(async () => {
    await prisma.$connect();
    for (const id of ['owner', 'other'])
      await prisma.user.create({
        data: { id, email: `${id}@example.invalid`, passwordHash: 'synthetic' },
      });
  });
  afterAll(async () => prisma.$disconnect());
  beforeEach(async () => {
    await prisma.tbankConnection.deleteMany();
    await prisma.tbankCatalogDiscovery.deleteMany();
    await prisma.tbankInstrument.deleteMany();
    await seed('owner');
    await seed('other');
    jest.resetAllMocks();
    upstream.findShares.mockResolvedValue([instrument]);
    upstream.findBonds.mockResolvedValue([
      { ...instrument, uid: 'bond', instrumentKind: 'INSTRUMENT_TYPE_BOND' },
    ]);
    service = createService();
  });
  it('persists normalized query identity and confirmed empty separately from unknown, across users and reopen', async () => {
    const cold = await service.read('owner', { ...query, query: ' sber ' });
    expect(cold.data).toBeNull();
    expect(cold.meta.availability).toBe('missing');
    await settled();
    await prisma.$disconnect();
    service = createService();
    const decrypt = jest.spyOn(tokens, 'decrypt');
    const warm = await service.read('other', query);
    expect(warm.data).toEqual([instrument]);
    expect(warm.meta.availability).toBe('complete');
    expect(warm.meta.sync.id).toBeNull();
    expect(decrypt).not.toHaveBeenCalled();
    expect(upstream.findShares).toHaveBeenCalledTimes(1);
    expect(
      (await service.read('other', { ...query, query: 'GAZP' }, true)).data,
    ).toBeNull();
    upstream.findShares.mockResolvedValue([]);
    await service.read('other', { ...query, query: 'GAZP' });
    await settled();
    expect(
      (await service.read('owner', { ...query, query: 'GAZP' })).data,
    ).toEqual([]);
    expect((await service.read('owner', query)).data).toEqual([instrument]);
    expect(upstream.findShares).toHaveBeenCalledTimes(2);
    decrypt.mockRestore();
    await service.read('owner', { ...query, instrumentType: 'bond' });
    await settled();
    expect(
      (await service.read('other', { ...query, instrumentType: 'bond' }))
        .data?.[0].uid,
    ).toBe('bond');
  });
  it('keeps selection ordering, deduplicates before limiting, and retains identity without enrichment or private fields', async () => {
    upstream.findShares.mockResolvedValue([
      { ...instrument, uid: 'unavailable', apiTradeAvailableFlag: false },
      {
        ...instrument,
        uid: 'preferred',
        apiTradeAvailableFlag: true,
        classCode: 'TQBR',
        quantity: '999',
        estimatedAmount: 'secret',
      },
      ...Array.from({ length: 40 }, (_, i) => ({
        ...instrument,
        uid: `other-${String(i).padStart(2, '0')}`,
        isin: undefined,
        ticker: 'OTHER',
      })),
      {
        ...instrument,
        uid: 'wrong-kind',
        instrumentKind: 'INSTRUMENT_TYPE_BOND',
      },
    ]);
    await service.read('owner', query);
    await settled();
    const saved = await service.read('other', query);
    expect(saved.data).toHaveLength(30);
    expect(saved.data?.[0].uid).toBe('preferred');
    expect(
      saved.data?.some((row) =>
        ['unavailable', 'wrong-kind'].includes(row.uid),
      ),
    ).toBe(false);
    expect(
      JSON.stringify(await prisma.tbankInstrumentSource.findMany()),
    ).not.toMatch(
      /quantity|estimatedAmount|secret|owner-revision|synthetic-owner/,
    );
    expect(await prisma.tbankInstrumentLookup.count()).toBe(0);
  });
  it('does not overwrite detailed metadata or its timestamp while refreshing discovery', async () => {
    const detailed = { ...instrument, sector: 'financial', lot: 10 };
    const fetchedAt = new Date('2026-01-01');
    await prisma.tbankInstrument.create({
      data: {
        provider: 'tbank',
        uid: 'uid',
        instrumentType: 'share',
        metadata: detailed,
        fetchedAt,
      },
    });
    await service.read('owner', query);
    await settled();
    const row = await prisma.tbankInstrument.findUniqueOrThrow({
      where: { provider_uid: { provider: 'tbank', uid: 'uid' } },
    });
    expect(row.metadata).toEqual(detailed);
    expect(row.fetchedAt).toEqual(fetchedAt);
  });
  it('coalesces twenty concurrent requests and forces with private handles and one upstream call', async () => {
    const blocked = deferred<CatalogInstrument[]>();
    upstream.findShares.mockReturnValue(blocked.promise);
    const reads = await Promise.all(
      Array.from({ length: 20 }, (_, i) =>
        service.read(i % 2 ? 'owner' : 'other', query),
      ),
    );
    await service.request('owner', 'owner-revision', query, true);
    await delay(10);
    expect(upstream.findShares).toHaveBeenCalledTimes(1);
    expect(new Set(reads.map((read) => read.meta.sync.id)).size).toBe(2);
    blocked.resolve([instrument]);
    await settled();
    expect(
      await prisma.tbankSyncWork.count({ where: { state: 'complete' } }),
    ).toBe(2);
  });
  it.each(['delete', 'replace'])(
    'discards late %s publication and does not borrow observer credentials',
    async (action) => {
      const blocked = deferred<CatalogInstrument[]>();
      upstream.findShares.mockReturnValueOnce(blocked.promise);
      await service.read('owner', query);
      await service.read('other', query);
      await delay(10);
      await prisma.tbankConnection.delete({ where: { userId: 'owner' } });
      if (action === 'replace')
        await prisma.tbankConnection.create({
          data: {
            userId: 'owner',
            revision: 'replacement',
            encryptedToken: tokens.encrypt('replacement'),
            connectedAt: new Date(),
          },
        });
      blocked.resolve([instrument]);
      await settled();
      expect(await prisma.tbankCatalogDiscovery.count()).toBe(0);
      expect(await prisma.tbankInstrument.count()).toBe(0);
      expect(
        (await service.read('other', query, true)).meta.sync.errorCode,
      ).toBe('TBANK_UNAVAILABLE');
      expect(upstream.findShares).toHaveBeenCalledTimes(1);
      await service.read('other', query);
      await settled();
      expect((await service.read('other', query, true)).data).toEqual([
        instrument,
      ]);
    },
  );
  it('preserves a successful result and its timestamp after a failed force without observe retry', async () => {
    await service.read('owner', query);
    await settled();
    const before = await repository.read(query);
    upstream.findShares.mockRejectedValue(new Error('outage'));
    await service.request('owner', 'owner-revision', query, true);
    await settled();
    for (let i = 0; i < 5; i++) await service.read('owner', query, true);
    expect(await repository.read(query)).toEqual(before);
    expect((await service.read('owner', query, true)).meta.sync.state).toBe(
      'failed',
    );
    expect(upstream.findShares).toHaveBeenCalledTimes(2);
    expect((await service.read('other', query)).meta.sync.errorCode).toBeNull();
  });
  it('isolates invalid credentials from shared empty lookup and another user', async () => {
    const blocked = deferred<CatalogInstrument[]>();
    upstream.findShares.mockReturnValueOnce(blocked.promise);
    await service.read('owner', query);
    await service.read('other', query);
    blocked.reject(new Error('INVALID_TBANK_TOKEN'));
    await settled();
    expect(await prisma.tbankCatalogDiscovery.count()).toBe(0);
    expect(
      (
        await prisma.tbankConnection.findUniqueOrThrow({
          where: { userId: 'owner' },
        })
      ).invalidToken,
    ).toBe(true);
    expect((await service.read('other', query, true)).meta.sync.errorCode).toBe(
      'TBANK_UNAVAILABLE',
    );
    await service.read('other', query);
    await settled();
    expect((await service.read('other', query, true)).data).toEqual([
      instrument,
    ]);
  });
  it('recovers abandoned work only on the next ordinary request and rejects superseded attempts', async () => {
    const abandoned = await repository.reserve(
      { userId: 'owner', revision: 'owner-revision' },
      query,
      'old-process',
      new Set(),
    );
    assert(abandoned);
    service = createService();
    await service.read('owner', query, true);
    await delay(10);
    expect(upstream.findShares).not.toHaveBeenCalled();
    await service.read('owner', query);
    await settled();
    expect(
      await repository.publish(
        {
          userId: 'owner',
          revision: 'owner-revision',
          id: abandoned.work.id,
          attemptId: abandoned.work.attemptId,
        },
        query,
        [],
        new Date(),
      ),
    ).toBe(false);
    expect((await service.read('owner', query, true)).data).toEqual([
      instrument,
    ]);
  });
  it('refreshes at the daily boundary while stale observe and fresh reads perform zero RPC', async () => {
    await service.read('owner', query);
    await settled();
    const now = Date.now();
    const clock = jest.spyOn(Date, 'now').mockReturnValue(now);
    try {
      await prisma.tbankCatalogDiscovery.updateMany({
        data: { fetchedAt: new Date(now - 86_399_000) },
      });
      await service.read('other', query);
      expect(upstream.findShares).toHaveBeenCalledTimes(1);
      await prisma.tbankCatalogDiscovery.updateMany({
        data: { fetchedAt: new Date(now - 86_400_000) },
      });
      expect((await service.read('other', query, true)).meta.freshness).toBe(
        'stale',
      );
      expect(upstream.findShares).toHaveBeenCalledTimes(1);
      await service.read('other', query);
      await settled();
      expect(upstream.findShares).toHaveBeenCalledTimes(2);
    } finally {
      clock.mockRestore();
    }
  });
});
function deferred<T>() {
  let resolve: (value: T) => void = () => {
    throw new Error('Uninitialized deferred');
  };
  let reject: (error: Error) => void = () => {
    throw new Error('Uninitialized deferred');
  };
  const promise = new Promise<T>((yes, no) => {
    resolve = yes;
    reject = no;
  });
  return { promise, resolve, reject };
}