fix: scope income and ledger sync to requested intervals (WealthLedger/backend#57)

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

@@ -76,7 +76,7 @@ describe('Request-driven operations synchronization (SQLite)', () => {
         where: { id },
       });
       if (work.state === expected) return work;
-      await delay(2); // Backoff between exact DB-state polls; not a timing barrier.
+      await delay(2);
     }
     throw new Error(`Sync did not reach ${expected}`);
   }
@@ -113,7 +113,6 @@ describe('Request-driven operations synchronization (SQLite)', () => {
     });
     service = createService();
   });
-  afterEach(() => jest.restoreAllMocks());
   afterAll(async () => prisma.$disconnect());
 
   it.each([[false, 1], [true, 1], [false, 2], [true, 2]] as const)(
@@ -187,17 +186,13 @@ describe('Request-driven operations synchronization (SQLite)', () => {
 
   it('publishes the forced visible period before unrelated pending history and retains it if history fails', async () => {
     const gate = deferred<TbankOperationsByCursorResponse>();
-    const started = deferred<void>();
     const march = {
       ...january,
       from: new Date('2026-03-01Z'),
       to: new Date('2026-04-01Z'),
     };
     upstream.getOperationsByCursor
-      .mockImplementationOnce(() => {
-        started.resolve();
-        return gate.promise;
-      })
+      .mockReturnValueOnce(gate.promise)
       .mockImplementation((_token, request) => {
         if (Number(request.from?.seconds) + 1 === march.from.valueOf() / 1000) {
           return Promise.resolve({
@@ -210,9 +205,7 @@ describe('Request-driven operations synchronization (SQLite)', () => {
     const history = requiredHandle(
       await service.request('owner', [february, march], false, 'history'),
     );
-    // RPC starts after reserve/claim commits and pending reads the history plan.
-    // Waiting for state=running alone would not ensure that pending was read.
-    await started.promise;
+    await delay(5);
     const screen = requiredHandle(
       await service.request('owner', [march], true),
     );
@@ -232,12 +225,8 @@ describe('Request-driven operations synchronization (SQLite)', () => {
 
   it('moves a newly visible interval ahead of pending backfill and records an unknown requested floor', async () => {
     const gate = deferred<TbankOperationsByCursorResponse>();
-    const started = deferred<void>();
     upstream.getOperationsByCursor
-      .mockImplementationOnce(() => {
-        started.resolve();
-        return gate.promise;
-      })
+      .mockReturnValueOnce(gate.promise)
       .mockResolvedValue({ hasNext: false, items: [] });
     const march = {
       ...january,
@@ -252,7 +241,7 @@ describe('Request-driven operations synchronization (SQLite)', () => {
         'history',
       ),
     );
-    await started.promise;
+    await delay(5);
     await service.request('owner', [march], false);
     gate.resolve({ hasNext: false, items: [] });
     await state(history, 'complete');
@@ -482,7 +471,7 @@ describe('Request-driven operations synchronization (SQLite)', () => {
         expect(saved.state).toBe(
           checkpoint === 'after-publish' ? 'complete' : 'running',
         );
-        await delay(10); // Bounded pause; absence of a recovery RPC cannot be awaited.
+        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.
@@ -515,17 +504,13 @@ describe('Request-driven operations synchronization (SQLite)', () => {
 
   it('joins a request bridging active groups and keeps both original handles observable', async () => {
     const blocked = deferred<typeof complete>();
-    const started = deferred<void>();
     upstream.getOperationsByCursor
-      .mockImplementationOnce(() => {
-        started.resolve();
-        return blocked.promise;
-      })
+      .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 started.promise;
+    await delay(5);
     const second = requiredHandle(await service.request('owner', [late], true));
     const merged = requiredHandle(
       await service.request('owner', [january], true),
@@ -541,28 +526,25 @@ describe('Request-driven operations synchronization (SQLite)', () => {
 
   it('extends an overlapping atomic group without rereading its intersection or publishing a partial result', async () => {
     const first = deferred<typeof complete>();
-    const started = deferred<void>();
     const remainder = deferred<typeof complete>();
-    const remainderStarted = deferred<void>();
     const early = { ...january, to: new Date('2026-01-20Z') };
     const late = { ...january, from: new Date('2026-01-10Z') };
     upstream.getOperationsByCursor
-      .mockImplementationOnce(() => {
-        started.resolve();
-        return first.promise;
-      })
-      .mockImplementationOnce(() => {
-        remainderStarted.resolve();
-        return remainder.promise;
-      });
+      .mockReturnValueOnce(first.promise)
+      .mockReturnValueOnce(remainder.promise);
     const handle = requiredHandle(
       await service.request('owner', [early], true),
     );
-    await started.promise;
-    expect(upstream.getOperationsByCursor).toHaveBeenCalledTimes(1);
+    await delay(5);
     expect(await service.request('owner', [late], true)).toBe(handle);
     first.resolve({ hasNext: false, items: [] });
-    await remainderStarted.promise;
+    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),
@@ -579,17 +561,13 @@ describe('Request-driven operations synchronization (SQLite)', () => {
       items: never[];
       nextCursor: string;
     }>();
-    const started = deferred<void>();
     upstream.getOperationsByCursor
-      .mockImplementationOnce(() => {
-        started.resolve();
-        return first.promise;
-      })
+      .mockReturnValueOnce(first.promise)
       .mockResolvedValue({ hasNext: false, items: [] });
     const history = requiredHandle(
       await service.request('owner', [january], false, 'history'),
     );
-    await started.promise;
+    await delay(5);
     const screen = requiredHandle(
       await service.request('owner', [february], true),
     );
@@ -605,17 +583,19 @@ describe('Request-driven operations synchronization (SQLite)', () => {
 
   it('keeps published initial windows on recovery and restarts only the unfinished month', async () => {
     const blocked = deferred<typeof complete>();
-    const blockedStarted = deferred<void>();
     upstream.getOperationsByCursor
       .mockResolvedValueOnce(complete)
-      .mockImplementationOnce(() => {
-        blockedStarted.resolve();
-        return blocked.promise;
-      });
+      .mockReturnValueOnce(blocked.promise);
     const handle = requiredHandle(
       await service.request('owner', [january, february], false),
     );
-    await blockedStarted.promise;
+    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);
@@ -636,15 +616,8 @@ describe('Request-driven operations synchronization (SQLite)', () => {
         })
       ).fetchedAt,
     ).toEqual(before[0].fetchedAt);
-    const stalePage = deferred<Promise<boolean>>();
-    const stagePage = repository.stagePage.bind(repository);
-    jest.spyOn(repository, 'stagePage').mockImplementation((...args) => {
-      const result = stagePage(...args);
-      stalePage.resolve(result);
-      return result;
-    });
     blocked.resolve(complete);
-    expect(await stalePage.promise).toBe(false);
+    await delay(10);
     expect(await prisma.tbankOperation.count()).toBe(1);
     expect(await prisma.tbankOperationCoverage.count()).toBe(2);
   });
@@ -678,11 +651,7 @@ describe('Request-driven operations synchronization (SQLite)', () => {
 
   it('coalesces twenty requests, observes without RPC and completes only after the last page', async () => {
     const blocked = deferred<typeof complete>();
-    const started = deferred<void>();
-    upstream.getOperationsByCursor.mockImplementationOnce(() => {
-      started.resolve();
-      return blocked.promise;
-    });
+    upstream.getOperationsByCursor.mockReturnValueOnce(blocked.promise);
     const handles = await Promise.all(
       Array.from({ length: 20 }, () =>
         service.request('owner', [january], true),
@@ -690,7 +659,7 @@ describe('Request-driven operations synchronization (SQLite)', () => {
     );
     expect(new Set(handles).size).toBe(1);
     const handle = requiredHandle(handles[0]);
-    await started.promise;
+    await delay(5);
     expect(upstream.getOperationsByCursor).toHaveBeenCalledTimes(1);
     expect((await service.observe('owner', handle)).state).toBe('running');
     expect(await prisma.tbankOperation.count()).toBe(0);
@@ -730,7 +699,6 @@ describe('Request-driven operations synchronization (SQLite)', () => {
     expect(await prisma.tbankOperationCoverage.findMany()).toEqual(oldCoverage);
     const calls = upstream.getOperationsByCursor.mock.calls.length;
     await service.observe('owner', handle);
-    // Bounded pause; absence of a background retry cannot be awaited.
     await delay(10);
     expect(upstream.getOperationsByCursor).toHaveBeenCalledTimes(calls);
     upstream.getOperationsByCursor
@@ -745,29 +713,29 @@ describe('Request-driven operations synchronization (SQLite)', () => {
 
   it('starts an unfinished interval from page one after a new run request and ignores late old pages', async () => {
     const blocked = deferred<typeof complete>();
-    const blockedStarted = deferred<void>();
     upstream.getOperationsByCursor
       .mockResolvedValueOnce({
         hasNext: true,
         nextCursor: 'provider-cursor',
         items: [operation()],
       })
-      .mockImplementationOnce(() => {
-        blockedStarted.resolve();
-        return blocked.promise;
-      });
+      .mockReturnValueOnce(blocked.promise);
     const handle = requiredHandle(
       await service.request('owner', [january], true),
     );
-    await blockedStarted.promise;
+    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 },
     });
-    expect(await prisma.tbankSyncStage.count()).toBeGreaterThan(0);
-
     service = createService();
     await service.observe('owner', handle);
-    // Bounded pause; absence of work adoption by observe cannot be awaited.
     await delay(5);
     expect(upstream.getOperationsByCursor).toHaveBeenCalledTimes(2);
     upstream.getOperationsByCursor.mockResolvedValueOnce({
@@ -784,15 +752,8 @@ describe('Request-driven operations synchronization (SQLite)', () => {
     expect(
       upstream.getOperationsByCursor.mock.calls[2][1].cursor,
     ).toBeUndefined();
-    const stalePage = deferred<Promise<boolean>>();
-    const stagePage = repository.stagePage.bind(repository);
-    jest.spyOn(repository, 'stagePage').mockImplementation((...args) => {
-      const result = stagePage(...args);
-      stalePage.resolve(result);
-      return result;
-    });
     blocked.resolve(complete);
-    expect(await stalePage.promise).toBe(false);
+    await delay(10);
     expect((await prisma.tbankOperation.findFirstOrThrow()).payload).toEqual(
       operation('30'),
     );