refactor: remove unreachable catalog producer and sync screen orchestration (WealthLedger/backend#62)

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

@@ -1,6 +1,5 @@
 import { ConfigService } from '@nestjs/config';
 import assert from 'node:assert/strict';
-import { setTimeout as delay } from 'node:timers/promises';
 
 import { TbankAccessService } from '../src/integrations/tbank/tbank-access.service';
 import { TbankTokenService } from '../src/integrations/tbank/tbank-token.service';
@@ -12,7 +11,7 @@ import {
 } from '../src/modules/investments/instruments/infrastructure/catalog-discovery.repository';
 import { PrismaService } from '../src/prisma/prisma.service';
 
-// Real transactions prove query coverage, private work ownership and late-publication fences.
+// Real transactions prove query normalization, confirmed-empty identity and cross-user isolation of shared reads.
 describe('shared catalog discovery (SQLite)', () => {
   const prisma = new PrismaService(
     new ConfigService({ DATABASE_URL: process.env.DATABASE_URL }),
@@ -30,7 +29,6 @@ describe('shared catalog discovery (SQLite)', () => {
       repository,
       access,
       upstream as never,
-      { event: jest.fn() } as never,
       { requestMany: jest.fn().mockResolvedValue([]) } as never,
     );
   let service: CatalogDiscoveryService;
@@ -51,14 +49,42 @@ describe('shared catalog discovery (SQLite)', () => {
         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);
+  // Writes a finished discovery directly, so the read path is tested without any producer.
+  async function seedCatalog(
+    query: string,
+    instruments: Array<{ payload: CatalogInstrument; uid: string }>,
+  ) {
+    for (const item of instruments) {
+      await prisma.tbankInstrument.create({
+        data: {
+          provider: 'tbank',
+          uid: item.uid,
+          instrumentType: 'share',
+          metadata: {},
+          fetchedAt: new Date(),
+        },
+      });
+      await prisma.tbankInstrumentSource.create({
+        data: {
+          provider: 'tbank',
+          instrumentUid: item.uid,
+          source: 'tbank',
+          kind: 'catalog',
+          rangeKey: 'share',
+          payload: item.payload,
+          fetchedAt: new Date(),
+        },
+      });
     }
-    throw new Error('Catalog work did not settle');
+    await prisma.tbankCatalogDiscovery.create({
+      data: {
+        provider: 'tbank',
+        instrumentType: 'share',
+        query,
+        instrumentUids: instruments.map((item) => item.uid),
+        fetchedAt: new Date(),
+      },
+    });
   }
   beforeAll(async () => {
     await prisma.$connect();
@@ -72,265 +98,41 @@ describe('shared catalog discovery (SQLite)', () => {
   beforeEach(async () => {
     await prisma.tbankConnection.deleteMany();
     await prisma.tbankCatalogDiscovery.deleteMany();
+    await prisma.tbankInstrumentSource.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 () => {
+  it('treats an unknown query as unavailable and a seeded query as available for another user', async () => {
     const cold = await service.read('owner', { ...query, query: ' sber ' });
     expect(cold.data).toBeNull();
-    expect(cold.meta.availability).toBe('initial-loading');
-    await settled();
-    await prisma.$disconnect();
-    service = createService();
-    const decrypt = jest.spyOn(tokens, 'decrypt');
+    expect(cold.meta.availability).toBe('temporarily-unavailable');
+    await seedCatalog('SBER', [{ uid: 'uid', payload: instrument }]);
     const warm = await service.read('other', query);
     expect(warm.data).toEqual([instrument]);
     expect(warm.meta.availability).toBe('available');
-    expect(warm.meta.completeness).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('keeps known discovery stable between refresh requests', 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('fresh');
-      expect(upstream.findShares).toHaveBeenCalledTimes(1);
-      await service.read('other', query);
-      expect(upstream.findShares).toHaveBeenCalledTimes(1);
-    } finally {
-      clock.mockRestore();
-    }
+  it('keeps a confirmed empty result distinct from an unknown query', async () => {
+    await seedCatalog('GAZP', []);
+    const empty = await service.read('owner', { ...query, query: 'GAZP' });
+    expect(empty.data).toEqual([]);
+    expect(empty.meta.availability).toBe('available');
+    const unknown = await service.read('owner', { ...query, query: 'LKOH' });
+    expect(unknown.data).toBeNull();
+  });
+  it('keeps the stored selection order and does not expose private fields', async () => {
+    const first = { ...instrument, uid: 'first', name: 'First' };
+    const second = { ...instrument, uid: 'second', name: 'Second' };
+    await seedCatalog('SBER', [
+      { uid: 'second', payload: second },
+      { uid: 'first', payload: first },
+    ]);
+    const result = await service.read('owner', query);
+    assert(result.data);
+    expect(result.data.map((item) => item.uid)).toEqual(['second', 'first']);
+    expect(result.data[0]).not.toHaveProperty('accountId');
   });
 });
-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 };
-}