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

test/tbank-income-local.e2e-spec.ts

@@ -297,6 +297,58 @@ describe('local income completeness and exact money (SQLite)', () => {
     expect(result.data?.calculatedAt).toBe('2025-02-01T00:00:00.000Z');
     expect(upstream.getOperationsByCursor).not.toHaveBeenCalled();
   });
+  it('ordinary covered income reads local data without membership or operations RPC', async () => {
+    await prisma.tbankConnection.update({
+      where: { userId: 'income-owner' },
+      data: { membershipAt: new Date('2020-01-01Z') },
+    });
+    const response = await get('a', undefined, false).expect(200);
+    const result = response.body as IncomeAnalyticsResourceDto;
+    expect(result.data).not.toBeNull();
+    expect(result.meta.completeness).toBe('complete');
+    expect(upstream.getOpenAccounts).not.toHaveBeenCalled();
+    expect(upstream.getOperationsByCursor).not.toHaveBeenCalled();
+  });
+  it('cold income imports only the missing requested interval', async () => {
+    await prisma.tbankConnection.update({
+      where: { userId: 'income-owner' },
+      data: { membershipAt: new Date('2020-01-01Z') },
+    });
+    let release: () => void = () => {
+      throw new Error('Upstream not initialized');
+    };
+    const blocked = new Promise<{ hasNext: boolean; items: never[] }>((resolve) => {
+      release = () => resolve({ hasNext: false, items: [] });
+    });
+    upstream.getOperationsByCursor.mockReturnValue(blocked);
+    const first = await get('a', 'from=2025-02-01&to=2025-02-28', false).expect(200);
+    const cold = first.body as IncomeAnalyticsResourceDto;
+    release();
+    expect(cold.data).toBeNull();
+    expect(cold.meta.completeness).toBe('incomplete');
+    let result: IncomeAnalyticsResourceDto = cold;
+    for (let attempt = 0; attempt < 100; attempt++) {
+      const response = await get('a', 'from=2025-02-01&to=2025-02-28').expect(200);
+      result = response.body as IncomeAnalyticsResourceDto;
+      if (result.meta.completeness === 'complete') break;
+      await new Promise((resolve) => setTimeout(resolve, 10));
+    }
+    expect(result.data).not.toBeNull();
+    expect(result.meta.completeness).toBe('complete');
+    expect(upstream.getOpenAccounts).not.toHaveBeenCalled();
+    expect(upstream.getOperationsByCursor.mock.calls).toEqual([
+      ['synthetic', expect.objectContaining({
+        accountId: 'a',
+        from: { seconds: '1738357199' },
+        to: { seconds: '1738368001' },
+      })],
+      ['synthetic', expect.objectContaining({
+        accountId: 'a',
+        from: { seconds: '1738367999' },
+        to: { seconds: '1740776401' },
+      })],
+    ]);
+  });
   it('returns while a cold upstream is blocked and validates access/period before scheduling', async () => {
     await get('absent', undefined, false).expect(404);
     await get('a', 'from=2025-02-31&to=2025-03-01', false).expect(400);

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

@@ -113,6 +113,50 @@ describe('persistent ledger HTTP and cursor (SQLite)', () => {
       }
     }
   });
+  it.each([
+    'from=2099-01-01',
+    'from=2025-03-01&to=2025-02-01',
+  ])('empty ledger range %s returns 200 without upstream RPC', async (query) => {
+    const response = await request(app.getHttpServer())
+      .get(`/api/v1/brokerage-accounts/a/operations?${query}`)
+      .set('Authorization', auth())
+      .expect(200);
+    const result = response.body as BrokerageOperationsResourceDto;
+    expect(result.data?.items).toEqual([]);
+    expect(upstream.getOperationsByCursor).not.toHaveBeenCalled();
+  });
+  it('cold ledger imports only the missing requested interval', async () => {
+    upstream.getOperationsByCursor.mockResolvedValue({
+      hasNext: false,
+      items: [],
+    });
+    await put().expect(200);
+    await request(app.getHttpServer())
+      .get('/api/v1/brokerage-accounts/a/operations?from=2025-02-01&to=2025-02-28')
+      .set('Authorization', auth())
+      .expect(200);
+    let complete = false;
+    for (let attempt = 0; attempt < 100; attempt++) {
+      const response = await request(app.getHttpServer())
+        .get('/api/v1/brokerage-accounts/a/operations?from=2025-02-01&to=2025-02-28&sync=observe')
+        .set('Authorization', auth())
+        .expect(200);
+      const result = response.body as BrokerageOperationsResourceDto;
+      complete = result.meta.completeness === 'complete';
+      if (complete) break;
+      await delay(10);
+    }
+    expect(complete).toBe(true);
+    expect(upstream.getOperationsByCursor).toHaveBeenCalledTimes(1);
+    expect(upstream.getOperationsByCursor).toHaveBeenCalledWith(
+      'synthetic',
+      expect.objectContaining({
+        accountId: 'a',
+        from: { seconds: '1738367999' },
+        to: { seconds: '1740787201' },
+      }),
+    );
+  });
   it('reads ordered pages with identical IDs/timestamps across accounts and no upstream RPC, including after reopen', async () => {
     let cursor: null | string = null;
     const ids: string[] = [];
@@ -409,7 +453,7 @@ describe('persistent ledger HTTP and cursor (SQLite)', () => {
     const result = (await get('sync=observe', 'a').expect(200))
       .body as BrokerageOperationsResourceDto;
     expect(result.meta.coverage?.historyBoundary).toBe('unknown');
-    expect(result.meta.coverage?.historyState).toBe('loading');
+    expect(result.meta.coverage?.historyState).toBe('missing');
   }, 30_000);
 
   it('returns cold missing while upstream is blocked, then confirms an empty interval; observe never retries failure', async () => {