diff --git a/src/subdomains/core/accounting/services/__tests__/ledger-mark.service.spec.ts b/src/subdomains/core/accounting/services/__tests__/ledger-mark.service.spec.ts index 0b29127d25..b5ed996d57 100644 --- a/src/subdomains/core/accounting/services/__tests__/ledger-mark.service.spec.ts +++ b/src/subdomains/core/accounting/services/__tests__/ledger-mark.service.spec.ts @@ -5,6 +5,7 @@ import { TestUtil } from 'src/shared/utils/test.util'; import { Util } from 'src/shared/utils/util'; import { createCustomLog } from 'src/subdomains/supporting/log/__mocks__/log.entity.mock'; import { Log } from 'src/subdomains/supporting/log/log.entity'; +import { FinancialLogAssetPrice } from 'src/subdomains/supporting/log/log.repository'; import { LogService } from 'src/subdomains/supporting/log/log.service'; import { LedgerMarkService } from '../ledger-mark.service'; @@ -60,6 +61,50 @@ function fakeGetFinancialLogs(allRows: Log[]): FakeGetFinancialLogs { }; } +/** Matches `LogService.getFinancialLogAssetPrices` signature. */ +type FakeGetFinancialLogAssetPrices = ( + from?: Date, + to?: Date, + limit?: number, + after?: number, +) => Promise; + +/** Expand a Log fixture into the SQL projection shape (same LEFT JOIN LATERAL semantics as the repository). */ +function logToAssetPrices(row: Log): FinancialLogAssetPrice[] { + // message::jsonb aborts the whole query on malformed JSON in production (fail-loud, see + // log.repository.ts) — this fake must throw too, not swallow it into an empty projection (Finding 4). + const assets = (JSON.parse(row.message) as { assets?: Record }).assets; + const entries = assets ? Object.entries(assets) : []; + + // LEFT JOIN LATERAL preserves the log row even when jsonb_each yields no rows (empty/absent assets): + // exactly one null-price placeholder row so logId-based overflow counting still sees this log row. + if (!entries.length) { + return [{ created: row.created, assetId: null, priceChf: null, logId: row.id }]; + } + + return entries.map(([assetIdKey, assetLog]) => { + const priceChf = (assetLog as { priceChf?: unknown })?.priceChf; + return { + created: row.created, + assetId: /^[0-9]+$/.test(assetIdKey) ? Number(assetIdKey) : null, + priceChf: typeof priceChf === 'number' && Number.isFinite(priceChf) ? priceChf : null, + logId: row.id, + }; + }); +} + +/** + * Fake `getFinancialLogAssetPrices`: filters/limits LOG rows first (same as SQL subquery LIMIT), then expands. + * Cursor is the underlying log id — not a flat result-row index. + */ +function fakeGetFinancialLogAssetPrices(allRows: Log[]): FakeGetFinancialLogAssetPrices { + const filterLogs = fakeGetFinancialLogs(allRows); + return async (from?: Date, to?: Date, limit?: number, after?: number): Promise => { + const logs = await filterLogs(from, false, to, limit, after); + return logs.flatMap(logToAssetPrices); + }; +} + describe('LedgerMarkService', () => { let service: LedgerMarkService; let logService: LogService; @@ -142,7 +187,15 @@ describe('LedgerMarkService', () => { await service.getMarkAtWidened(5, asOf, 90); - expect(spy).toHaveBeenCalledWith(Util.daysBefore(90, asOf), true, asOf, Config.ledger.markPreloadMaxRows + 1); + // trailing undefined: the first page carries no cursor yet — asserted explicitly so a + // dropped or reordered cursor argument still fails this test. + expect(spy).toHaveBeenCalledWith( + Util.daysBefore(90, asOf), + true, + asOf, + Config.ledger.markPreloadMaxRows + 1, + undefined, + ); }); it('never returns a mark created after asOf', async () => { @@ -207,12 +260,14 @@ describe('LedgerMarkService', () => { it('returns the priceChf of the latest mark ≤ bookingDate (stage 2)', async () => { jest - .spyOn(logService, 'getFinancialLogs') - .mockResolvedValue([ - financialLog(new Date('2026-06-01'), { '5': { priceChf: 50000 } }), - financialLog(new Date('2026-06-02'), { '5': { priceChf: 51000 } }), - financialLog(new Date('2026-06-03'), { '5': { priceChf: 52000 } }), - ]); + .spyOn(logService, 'getFinancialLogAssetPrices') + .mockImplementation( + fakeGetFinancialLogAssetPrices([ + financialLog(new Date('2026-06-01'), { '5': { priceChf: 50000 } }), + financialLog(new Date('2026-06-02'), { '5': { priceChf: 51000 } }), + financialLog(new Date('2026-06-03'), { '5': { priceChf: 52000 } }), + ]), + ); const cache = await service.preload(new Date('2026-06-01'), new Date('2026-06-03')); @@ -222,8 +277,10 @@ describe('LedgerMarkService', () => { it('returns undefined when no log row ≤ bookingDate exists (stage 3 → needsMark)', async () => { jest - .spyOn(logService, 'getFinancialLogs') - .mockResolvedValue([financialLog(new Date('2026-06-05'), { '5': { priceChf: 50000 } })]); + .spyOn(logService, 'getFinancialLogAssetPrices') + .mockImplementation( + fakeGetFinancialLogAssetPrices([financialLog(new Date('2026-06-05'), { '5': { priceChf: 50000 } })]), + ); const cache = await service.preload(new Date('2026-06-05'), new Date('2026-06-05')); @@ -232,8 +289,10 @@ describe('LedgerMarkService', () => { it('returns undefined when a log row exists but its assets JSON lacks the assetId (Minor R5-5)', async () => { jest - .spyOn(logService, 'getFinancialLogs') - .mockResolvedValue([financialLog(new Date('2026-06-01'), { '7': { priceChf: 1.0 } })]); + .spyOn(logService, 'getFinancialLogAssetPrices') + .mockImplementation( + fakeGetFinancialLogAssetPrices([financialLog(new Date('2026-06-01'), { '7': { priceChf: 1.0 } })]), + ); const cache = await service.preload(new Date('2026-06-01'), new Date('2026-06-01')); @@ -242,24 +301,51 @@ describe('LedgerMarkService', () => { it('skips non-finite priceChf entries (no phantom 0 mark)', async () => { jest - .spyOn(logService, 'getFinancialLogs') - .mockResolvedValue([financialLog(new Date('2026-06-01'), { '5': { priceChf: NaN } })]); + .spyOn(logService, 'getFinancialLogAssetPrices') + .mockImplementation( + fakeGetFinancialLogAssetPrices([financialLog(new Date('2026-06-01'), { '5': { priceChf: NaN } })]), + ); const cache = await service.preload(new Date('2026-06-01'), new Date('2026-06-01')); expect(cache.getMarkAt(5, new Date('2026-06-01'))).toBeUndefined(); }); - it('never throws on malformed message JSON (defensive parse)', async () => { + it('throws on malformed message JSON (fail-loud, matches `message::jsonb` in production SQL)', async () => { jest - .spyOn(logService, 'getFinancialLogs') - .mockResolvedValue([createCustomLog({ created: new Date('2026-06-01'), message: 'not-json' })]); + .spyOn(logService, 'getFinancialLogAssetPrices') + .mockImplementation( + fakeGetFinancialLogAssetPrices([createCustomLog({ created: new Date('2026-06-01'), message: 'not-json' })]), + ); + + await expect(service.preload(new Date('2026-06-01'), new Date('2026-06-01'))).rejects.toThrow(); + }); + + it('excludes a priceChf that is a JSON string, not a JSON number (matches the old Number.isFinite gate)', async () => { + const row = financialLog(new Date('2026-06-01'), { + '5': { priceChf: '1.25' as unknown as number }, + }); + + jest.spyOn(logService, 'getFinancialLogAssetPrices').mockImplementation(fakeGetFinancialLogAssetPrices([row])); const cache = await service.preload(new Date('2026-06-01'), new Date('2026-06-01')); expect(cache.getMarkAt(5, new Date('2026-06-01'))).toBeUndefined(); }); + it('excludes a non-numeric asset key without aborting the rest of the projection', async () => { + const row = financialLog(new Date('2026-06-01'), { + abc: { priceChf: 10 }, + '5': { priceChf: 20 }, + }); + + jest.spyOn(logService, 'getFinancialLogAssetPrices').mockImplementation(fakeGetFinancialLogAssetPrices([row])); + + const cache = await service.preload(new Date('2026-06-01'), new Date('2026-06-01')); + + expect(cache.getMarkAt(5, new Date('2026-06-01'))).toBe(20); // numeric key still projected despite the sibling non-numeric key + }); + it('uses dailySample when the span exceeds the threshold (bounded preload)', async () => { const spy = jest .spyOn(logService, 'getFinancialLogs') @@ -272,39 +358,88 @@ describe('LedgerMarkService', () => { true, new Date('2026-06-10'), Config.ledger.markPreloadMaxRows + 1, + undefined, // first page carries no cursor ); }); it('uses the full minute-tick for fresh windows within the threshold', async () => { - const spy = jest - .spyOn(logService, 'getFinancialLogs') - .mockResolvedValue([financialLog(new Date('2026-06-01'), { '5': { priceChf: 50000 } })]); + const spy = jest.spyOn(logService, 'getFinancialLogAssetPrices').mockResolvedValue([ + { + created: new Date('2026-06-01'), + assetId: 5, + priceChf: 50000, + logId: 1, + }, + ]); await service.preload(new Date('2026-06-01'), new Date('2026-06-01T06:00:00Z')); // < 2 days expect(spy).toHaveBeenCalledWith( new Date('2026-06-01'), - false, new Date('2026-06-01T06:00:00Z'), Config.ledger.markPreloadMaxRows + 1, + undefined, // first page carries no cursor ); }); it('passes to and limit (maxRows + 1) on the preload trigger read', async () => { const to = new Date('2026-06-02'); - const spy = jest - .spyOn(logService, 'getFinancialLogs') - .mockResolvedValue([financialLog(new Date('2026-06-01'), { '5': { priceChf: 50000 } })]); + const spy = jest.spyOn(logService, 'getFinancialLogAssetPrices').mockResolvedValue([ + { + created: new Date('2026-06-01'), + assetId: 5, + priceChf: 50000, + logId: 1, + }, + ]); await service.preload(new Date('2026-06-01'), to); // Upper bound and row cap are enforced in SQL (no post-load JS filter); +1 keeps overflow detectable. - expect(spy).toHaveBeenCalledWith(new Date('2026-06-01'), false, to, Config.ledger.markPreloadMaxRows + 1); + expect(spy).toHaveBeenCalledWith(new Date('2026-06-01'), to, Config.ledger.markPreloadMaxRows + 1, undefined); }); - // §5.2 step 3 pagination backstop: when the first bounded read returns more than markPreloadMaxRows rows the service - // continues via keyset pages over id (created resolved in-DB). With markPreloadMaxRows=1 the first read (2 rows) - // trips the backstop; the probe's first maxRows rows are reused as page 1. + // Projection → same mark map semantics as the former full-Log path (assets, prices, order). + it('buildMarkMap from the asset-price projection yields the same marks as full log rows would', async () => { + const t0 = new Date('2026-06-01T00:00:00Z'); + const t1 = new Date('2026-06-01T01:00:00Z'); + const projection: FinancialLogAssetPrice[] = [ + { created: t0, assetId: 5, priceChf: 50000, logId: 10 }, + { created: t0, assetId: 6, priceChf: 1.5, logId: 10 }, + { created: t1, assetId: 5, priceChf: 51000, logId: 11 }, + ]; + + jest.spyOn(logService, 'getFinancialLogAssetPrices').mockResolvedValue(projection); + + const cache = await service.preload(t0, t1); + + expect(cache.getMarkAt(5, t0)).toBe(50000); + expect(cache.getMarkAt(6, t0)).toBe(1.5); + expect(cache.getMarkAt(5, t1)).toBe(51000); + expect(cache.getMarkAt(5, new Date('2026-06-01T00:30:00Z'))).toBe(50000); + }); + + it('omits assets with missing or non-finite projected prices (no 0-mark)', async () => { + const t = new Date('2026-06-01T00:00:00Z'); + // Repo/fake null non-finite / unusable fields; inject null placeholders so buildMarkMap's skip is covered. + jest.spyOn(logService, 'getFinancialLogAssetPrices').mockResolvedValue([ + { created: t, assetId: 5, priceChf: 42, logId: 1 }, + { created: t, assetId: 6, priceChf: null, logId: 1 }, + { created: t, assetId: null, priceChf: 7, logId: 1 }, + // asset 8 absent entirely → no mark + ]); + + const cache = await service.preload(t, t); + + expect(cache.getMarkAt(5, t)).toBe(42); + expect(cache.getMarkAt(6, t)).toBeUndefined(); + expect(cache.getMarkAt(7, t)).toBeUndefined(); + expect(cache.getMarkAt(8, t)).toBeUndefined(); + }); + + // §5.2 step 3 pagination backstop: when the first bounded read returns more than markPreloadMaxRows log rows the + // service continues via keyset pages over log id. With markPreloadMaxRows=1 the first read (2 log rows) trips the + // backstop; the probe's first maxRows complete log groups are reused as page 1. describe('pagination backstop (rows > markPreloadMaxRows)', () => { let pagedService: LedgerMarkService; @@ -327,11 +462,13 @@ describe('LedgerMarkService', () => { const w1b = financialLog(new Date('2026-06-01T01:00:00Z'), { '5': { priceChf: 51000 } }); const w2 = financialLog(new Date('2026-06-01T02:00:00Z'), { '5': { priceChf: 52000 } }); - const spy = jest.spyOn(logService, 'getFinancialLogs').mockImplementation(fakeGetFinancialLogs([w1a, w1b, w2])); + const spy = jest + .spyOn(logService, 'getFinancialLogAssetPrices') + .mockImplementation(fakeGetFinancialLogAssetPrices([w1a, w1b, w2])); const cache = await pagedService.preload(new Date('2026-06-01T00:00:00Z'), new Date('2026-06-01T03:00:00Z')); - // the continuation windows were used (>1 getFinancialLogs call beyond the trigger read) + // the continuation windows were used (>1 projection call beyond the trigger read) expect(spy.mock.calls.length).toBeGreaterThan(1); // all three marks made it into the cache built from the paginated rows expect(cache.getMarkAt(5, new Date('2026-06-01T00:30:00Z'))).toBe(50000); @@ -344,7 +481,7 @@ describe('LedgerMarkService', () => { const r1 = financialLog(new Date('2026-06-01T00:00:00Z'), { '5': { priceChf: 50000 } }); const r2 = financialLog(new Date('2026-06-01T01:00:00Z'), { '5': { priceChf: 51000 } }); - jest.spyOn(logService, 'getFinancialLogs').mockImplementation(fakeGetFinancialLogs([r1, r2])); + jest.spyOn(logService, 'getFinancialLogAssetPrices').mockImplementation(fakeGetFinancialLogAssetPrices([r1, r2])); const cache = await pagedService.preload(new Date('2026-06-01T00:00:00Z'), new Date('2026-06-01T03:00:00Z')); @@ -372,7 +509,9 @@ describe('LedgerMarkService', () => { message: JSON.stringify({ assets: { '6': { priceChf: 51000 } } }), }); - jest.spyOn(logService, 'getFinancialLogs').mockImplementation(fakeGetFinancialLogs([rowA, rowB])); + jest + .spyOn(logService, 'getFinancialLogAssetPrices') + .mockImplementation(fakeGetFinancialLogAssetPrices([rowA, rowB])); const cache = await pagedService.preload(t0, new Date('2026-06-01T03:00:00Z')); @@ -387,8 +526,8 @@ describe('LedgerMarkService', () => { const to = new Date('2026-06-01T03:00:00Z'); const spy = jest - .spyOn(logService, 'getFinancialLogs') - .mockImplementation(fakeGetFinancialLogs([inRange, alsoInRange, afterTo])); + .spyOn(logService, 'getFinancialLogAssetPrices') + .mockImplementation(fakeGetFinancialLogAssetPrices([inRange, alsoInRange, afterTo])); const cache = await pagedService.preload(new Date('2026-06-01T00:00:00Z'), to); @@ -397,14 +536,16 @@ describe('LedgerMarkService', () => { // too-late row must not leak into the cache (lookup at/after its created still shows the last in-range mark) expect(cache.getMarkAt(5, afterTo.created)).toBe(51000); expect(cache.getMarkAt(5, afterTo.created)).not.toBe(99999); - // every call passes the same upper bound + // every call passes the same upper bound (arg index 1 = `to` on getFinancialLogAssetPrices) for (const call of spy.mock.calls) { - expect(call[2]).toEqual(to); + expect(call[1]).toEqual(to); } }); it('returns an empty cache when no financial logs fall in the window', async () => { - const spy = jest.spyOn(logService, 'getFinancialLogs').mockImplementation(fakeGetFinancialLogs([])); + const spy = jest + .spyOn(logService, 'getFinancialLogAssetPrices') + .mockImplementation(fakeGetFinancialLogAssetPrices([])); const cache = await pagedService.preload(new Date('2026-06-01T00:00:00Z'), new Date('2026-06-01T03:00:00Z')); @@ -412,6 +553,32 @@ describe('LedgerMarkService', () => { expect(cache.getMarkAt(5, new Date('2026-06-01T00:30:00Z'))).toBeUndefined(); }); + // Multi-asset log rows: overflow/page boundaries must cut on logId groups, never mid-snapshot by array index. + it('keeps every asset of a multi-asset log when maxRows=1 (no mid-group cut on logId)', async () => { + const r1 = financialLog(new Date('2026-06-01T00:00:00Z'), { + '5': { priceChf: 1 }, + '6': { priceChf: 2 }, + }); + const r2 = financialLog(new Date('2026-06-01T01:00:00Z'), { '7': { priceChf: 3 } }); + + const spy = jest + .spyOn(logService, 'getFinancialLogAssetPrices') + .mockImplementation(fakeGetFinancialLogAssetPrices([r1, r2])); + + const cache = await pagedService.preload(new Date('2026-06-01T00:00:00Z'), new Date('2026-06-01T03:00:00Z')); + + // both logs fully present — slicing the flat projection at maxRows=1 would have dropped asset 6 + expect(cache.getMarkAt(5, new Date('2026-06-01T00:30:00Z'))).toBe(1); + expect(cache.getMarkAt(6, new Date('2026-06-01T00:30:00Z'))).toBe(2); + expect(cache.getMarkAt(7, new Date('2026-06-01T01:30:00Z'))).toBe(3); + expect(spy.mock.calls.length).toBeGreaterThan(1); + // continuation cursor is a log id (number), not a flat result index + for (const call of spy.mock.calls) { + const afterArg = call[3]; + if (afterArg !== undefined) expect(typeof afterArg).toBe('number'); + } + }); + // §5.2 precision fix: log.created is timestamp(6) in Postgres (microsecond precision) but JS `Date` // only carries milliseconds - reading a row truncates e.g. ...841802 -> ...841. A (created, id)-Date // cursor sent back to the DB compares this truncated value against the SAME row's full-precision @@ -433,7 +600,9 @@ describe('LedgerMarkService', () => { // pagedService here = the pagination-backstop describe-block's service instance with // markPreloadMaxRows = 1 (see that block's beforeEach) - reuse it, do not rebuild a separate module. - const spy = jest.spyOn(logService, 'getFinancialLogs').mockImplementation(fakeGetFinancialLogs(rows)); + const spy = jest + .spyOn(logService, 'getFinancialLogAssetPrices') + .mockImplementation(fakeGetFinancialLogAssetPrices(rows)); const cache = await pagedService.preload(new Date(sameMs), new Date(sameMs)); @@ -442,10 +611,9 @@ describe('LedgerMarkService', () => { expect(cache.getMarkAt(7, new Date(sameMs))).toBe(3); expect(spy.mock.calls.length).toBeLessThan(10); // terminates - no infinite loop - // structural guarantee: the cursor argument passed to getFinancialLogs is always a plain id - // (number), never a Date - so there is no JS-truncated timestamp for the DB to compare at all. + // structural guarantee: the cursor argument is always a plain log id (number), never a Date for (const call of spy.mock.calls) { - const afterArg = call[4]; + const afterArg = call[3]; if (afterArg !== undefined) expect(typeof afterArg).toBe('number'); } }); @@ -480,7 +648,9 @@ describe('LedgerMarkService', () => { const r3 = financialLog(sameCreated, { '7': { priceChf: 3 } }); const r4 = financialLog(new Date('2026-06-01T18:00:00Z'), { '8': { priceChf: 4 } }); - const spy = jest.spyOn(logService, 'getFinancialLogs').mockImplementation(fakeGetFinancialLogs([r1, r2, r3, r4])); + const spy = jest + .spyOn(logService, 'getFinancialLogAssetPrices') + .mockImplementation(fakeGetFinancialLogAssetPrices([r1, r2, r3, r4])); const cache = await pagedService.preload(batchStart, to); @@ -493,7 +663,7 @@ describe('LedgerMarkService', () => { expect(spy.mock.calls.length).toBeLessThan(10); }); - // Overflow reuse: probe reads maxRows+1, reuses first maxRows as page 1, continues from that last id. + // Overflow reuse: probe reads maxRows+1, reuses first maxRows complete log groups as page 1, continues from that last logId. // Without reuse: 1 probe + 3 pages = 4 calls. With reuse: 1 probe + 2 continuation pages = 3. it('reuses the overflow probe as page 1 (no double-read of the first maxRows rows)', async () => { const batchStart = new Date('2026-06-01T00:00:00Z'); @@ -505,8 +675,8 @@ describe('LedgerMarkService', () => { const r5 = financialLog(new Date('2026-06-01T05:00:00Z'), { '5': { priceChf: 50 } }); const spy = jest - .spyOn(logService, 'getFinancialLogs') - .mockImplementation(fakeGetFinancialLogs([r1, r2, r3, r4, r5])); + .spyOn(logService, 'getFinancialLogAssetPrices') + .mockImplementation(fakeGetFinancialLogAssetPrices([r1, r2, r3, r4, r5])); const cache = await pagedService.preload(batchStart, to); @@ -521,8 +691,44 @@ describe('LedgerMarkService', () => { }); }); + // Finding 1+2: a log row with no usable prices must still occupy a logId slot so overflow/pagination continue + // past it — otherwise later valid marks are silently dropped when maxRows=1. + describe('pagination continues past a log row with no usable prices (Finding 1+2 regression)', () => { + let pagedService: LedgerMarkService; + + beforeEach(async () => { + const module: TestingModule = await Test.createTestingModule({ + providers: [ + TestUtil.provideConfig({ + ledger: { markPreloadMaxRows: 1, markPreloadDailySampleThresholdDays: 2 }, + }), + LedgerMarkService, + { provide: LogService, useValue: logService }, + ], + }).compile(); + pagedService = module.get(LedgerMarkService); + }); + + it('keeps every valid mark when an unusable-price row sits between two valid rows', async () => { + const r1 = financialLog(new Date('2026-06-01T00:00:00Z'), { '5': { priceChf: 1 } }); + const unusable = financialLog(new Date('2026-06-01T01:00:00Z'), {}); // no usable price at all + const r3 = financialLog(new Date('2026-06-01T02:00:00Z'), { '5': { priceChf: 3 } }); + + const spy = jest + .spyOn(logService, 'getFinancialLogAssetPrices') + .mockImplementation(fakeGetFinancialLogAssetPrices([r1, unusable, r3])); + + const cache = await pagedService.preload(new Date('2026-06-01T00:00:00Z'), new Date('2026-06-01T03:00:00Z')); + + expect(cache.getMarkAt(5, new Date('2026-06-01T00:30:00Z'))).toBe(1); + expect(cache.getMarkAt(5, new Date('2026-06-01T02:30:00Z'))).toBe(3); // r3 must not be silently dropped + expect(spy.mock.calls.length).toBeGreaterThan(1); // pagination actually continued past the unusable row + }); + }); + // dailySample=true pagination path: span > threshold so both getFinancialLogs branches' cursor logic is exercised - // under overflow (maxRows small enough that multi-page keyset runs). + // under overflow (maxRows small enough that multi-page keyset runs). Stays on getFinancialLogs (no dailySample + // parameter on getFinancialLogAssetPrices) — rare long-window path only. describe('pagination with dailySample=true (span > markPreloadDailySampleThresholdDays)', () => { let pagedService: LedgerMarkService; diff --git a/src/subdomains/core/accounting/services/ledger-mark.service.ts b/src/subdomains/core/accounting/services/ledger-mark.service.ts index 4aea62f446..fda1173e47 100644 --- a/src/subdomains/core/accounting/services/ledger-mark.service.ts +++ b/src/subdomains/core/accounting/services/ledger-mark.service.ts @@ -3,6 +3,7 @@ import { Config } from 'src/config/config'; import { Util } from 'src/shared/utils/util'; import { FinanceLog } from 'src/subdomains/supporting/log/dto/log.dto'; import { Log } from 'src/subdomains/supporting/log/log.entity'; +import { FinancialLogAssetPrice } from 'src/subdomains/supporting/log/log.repository'; import { LogService } from 'src/subdomains/supporting/log/log.service'; interface MarkPoint { @@ -80,6 +81,8 @@ export class LedgerMarkService { } // bounded, memoized youngest-mark map (≤ now). Ascending by created → the last finite write per asset wins → youngest. + // Stays on full getFinancialLogs (not the price projection): does not use buildMarkMap / preload pagination and only + // needs a flat Map over a short daily-sampled window. private async getLatestMarks(): Promise> { const now = Date.now(); if (this.latestMarks && now - this.latestMarks.loadedAt < LATEST_MARK_TTL_MS) return this.latestMarks.map; @@ -138,48 +141,117 @@ export class LedgerMarkService { /** * Bounded preload (§5.2, Hard Constraint #4): always limited by (batchStartDate, to) and maxRows. * Order is fixed — dailySample decision FIRST (avoids loading the full minute-tick), THEN upper-bound - * trimming, THEN the maxRows pagination backstop (keyset over id; created resolved in-DB). + * trimming, THEN the maxRows pagination backstop (keyset over log id; created resolved in-DB). + * + * Hot path (dailySample=false): SQL projects priceChf only (getFinancialLogAssetPrices). + * Rare long-window path (dailySample=true): still getFinancialLogs + local expansion to the same projection type. */ async preload(batchStartDate: Date, to: Date): Promise { const spanDays = Util.daysDiff(batchStartDate, to); const dailySample = spanDays > Config.ledger.markPreloadDailySampleThresholdDays; const maxRows = this.getMarkPreloadMaxRows(); - // +1 so probeRows.length > maxRows can still detect overflow when SQL already caps at maxRows - const probeRows = await this.logService.getFinancialLogs(batchStartDate, dailySample, to, maxRows + 1); + // +1 so unique log-id count > maxRows can still detect overflow when SQL already caps at maxRows log rows + const probeRows = await this.loadAssetPrices(batchStartDate, to, dailySample, maxRows + 1); const rows = - probeRows.length > maxRows - ? await this.paginate(batchStartDate, to, dailySample, probeRows.slice(0, maxRows)) + this.uniqueLogIds(probeRows).length > maxRows + ? await this.paginate(batchStartDate, to, dailySample, this.takeCompleteLogGroups(probeRows, maxRows)) : probeRows; return new LedgerMarkCache(this.buildMarkMap(rows)); } - // Keyset pages over id; never load everything into one heap (§5.2 step 3). - // `firstPage` reuses the rows preload() already read via the overflow probe (the same maxRows-sized - // first page a from-scratch pagination would produce, since both share the same filters/order/limit= - // maxRows and deterministic ORDER BY created ASC, id ASC) — so the probe read is not thrown away and - // re-fetched. Halves the data read on overflow, result set stays identical to full re-pagination. - private async paginate(batchStartDate: Date, to: Date, dailySample: boolean, firstPage: Log[]): Promise { + // Keyset pages over log id; never load everything into one heap (§5.2 step 3). + // Overflow/page sizes are measured in distinct logId groups (one FinancialDataLog snapshot), not flattened + // asset-result length — slicing by array index would cut mid-snapshot and drop assets silently. + // `firstPage` reuses the complete log groups preload() already read via the overflow probe. + private async paginate( + batchStartDate: Date, + to: Date, + dailySample: boolean, + firstPage: FinancialLogAssetPrice[], + ): Promise { const maxRows = this.getMarkPreloadMaxRows(); - const result: Log[] = [...firstPage]; - let after: number | undefined = firstPage[firstPage.length - 1]?.id; + const result: FinancialLogAssetPrice[] = [...firstPage]; + const firstPageLogIds = this.uniqueLogIds(firstPage); + let after: number | undefined = firstPageLogIds[firstPageLogIds.length - 1]; - // Keyset continuation: each page starts strictly after the last returned id. + // Keyset continuation: each page starts strictly after the last returned log id. while (true) { - const window = await this.logService.getFinancialLogs(batchStartDate, dailySample, to, maxRows, after); + const window = await this.loadAssetPrices(batchStartDate, to, dailySample, maxRows, after); if (!window.length) break; result.push(...window); - if (window.length < maxRows) break; + const windowLogIds = this.uniqueLogIds(window); + if (windowLogIds.length < maxRows) break; - after = window[window.length - 1].id; + after = windowLogIds[windowLogIds.length - 1]; } return result; } + // Hot path: SQL projection. dailySample path: full logs (day-group SQL not reimplemented) expanded locally. + private async loadAssetPrices( + from: Date, + to: Date, + dailySample: boolean, + limit?: number, + after?: number, + ): Promise { + if (dailySample) { + const logs = await this.logService.getFinancialLogs(from, true, to, limit, after); + return logs.flatMap((row) => this.rowToAssetPrices(row)); + } + return this.logService.getFinancialLogAssetPrices(from, to, limit, after); + } + + // Expand one Log into the same projection shape as getFinancialLogAssetPrices (dailySample adapter only). + // Mirrors the repository's LEFT JOIN LATERAL: every log row must yield at least one result row so that + // uniqueLogIds/overflow-detection below count log rows actually read, not just the ones carrying a usable + // mark (see PR review Finding 1+2) — an empty/absent `assets` object still emits one all-null placeholder + // row, and each present asset key emits its own row with assetId/priceChf nulled per-field when unusable + // (non-numeric key / non-finite price), rather than being skipped. + private rowToAssetPrices(row: Log): FinancialLogAssetPrice[] { + const assets = this.parseAssets(row.message); + const entries = assets ? Object.entries(assets) : []; + + if (!entries.length) { + return [{ created: row.created, assetId: null, priceChf: null, logId: row.id }]; + } + + return entries.map(([assetIdKey, assetLog]) => { + const priceChf = assetLog?.priceChf; + return { + created: row.created, + assetId: /^[0-9]+$/.test(assetIdKey) ? Number(assetIdKey) : null, + priceChf: typeof priceChf === 'number' && Number.isFinite(priceChf) ? priceChf : null, + logId: row.id, + }; + }); + } + + // Distinct logIds in first-seen order (SQL keeps all assets of one log contiguous). + // logId is present on every projection row, including null-price placeholders, so this counts log rows + // actually read — not only those that carried a usable mark (Finding 1+2). + private uniqueLogIds(rows: FinancialLogAssetPrice[]): number[] { + const ids: number[] = []; + const seen = new Set(); + for (const row of rows) { + if (seen.has(row.logId)) continue; + seen.add(row.logId); + ids.push(row.logId); + } + return ids; + } + + // Keep every projected asset belonging to the first `maxLogRows` distinct logIds (no mid-group cut). + private takeCompleteLogGroups(rows: FinancialLogAssetPrice[], maxLogRows: number): FinancialLogAssetPrice[] { + const allowed = new Set(this.uniqueLogIds(rows).slice(0, maxLogRows)); + return rows.filter((row) => allowed.has(row.logId)); + } + // Fail loud on a non-positive / non-integer markPreloadMaxRows (e.g. LEDGER_MARK_PRELOAD_MAX_ROWS=0 or // a broken env parse): LIMIT 0 / empty first page would otherwise silently build an empty cache. private getMarkPreloadMaxRows(): number { @@ -190,26 +262,22 @@ export class LedgerMarkService { return value; } - private buildMarkMap(rows: Log[]): Map { + private buildMarkMap(rows: FinancialLogAssetPrice[]): Map { const marks = new Map(); for (const row of rows) { - // tolerate parse/shape issues defensively — never throw, mirrors log-job getJsonValue - const assets = this.parseAssets(row.message); - if (!assets) continue; - - for (const [assetIdKey, assetLog] of Object.entries(assets)) { - const priceChf = assetLog?.priceChf; - if (!Number.isFinite(priceChf)) continue; - - const assetId = +assetIdKey; - const points = marks.get(assetId) ?? []; - points.push({ created: row.created, priceChf }); - marks.set(assetId, points); - } + // Repo / rowToAssetPrices keep a row per read log even without a usable mark (assetId/priceChf + // null) so overflow detection and keyset pagination count log rows correctly — skip those here. + // Number.isFinite is kept as a second line of defence: the repository already nulls NaN/Infinity, + // but a mark of NaN would silently corrupt a valuation, so it must not depend on one guard alone. + if (row.assetId == null || !Number.isFinite(row.priceChf)) continue; + + const points = marks.get(row.assetId) ?? []; + points.push({ created: row.created, priceChf: row.priceChf }); + marks.set(row.assetId, points); } - // rows arrive ascending by created (getFinancialLogs order); keep lists sorted for binary search + // rows arrive ascending by created (repository order); keep lists sorted for binary search for (const points of marks.values()) { points.sort((a, b) => a.created.getTime() - b.created.getTime()); } diff --git a/src/subdomains/supporting/log/log.repository.ts b/src/subdomains/supporting/log/log.repository.ts index fc55654bb8..6be2809ef7 100644 --- a/src/subdomains/supporting/log/log.repository.ts +++ b/src/subdomains/supporting/log/log.repository.ts @@ -19,6 +19,27 @@ import { MAX_VALIDITY_SWEEP_ROWS, } from './log.entity'; +/** One asset priceChf projected from a FinancialDataLog snapshot (no full message JSON). */ +export interface FinancialLogAssetPrice { + created: Date; + /** + * null when the JSON key in `assets` is not a plain non-negative-integer string (`^[0-9]+$`) — + * kept as a row with assetId=null rather than aborting the whole query via a failing `::int` cast. + */ + assetId: number | null; + /** + * null when this row carries no usable price: `assets` was empty/absent for the log row (one + * placeholder row per log row), the JSON value at `priceChf` was not a JSON number (e.g. the string + * "1.25"), or the numeric value was NaN/Infinity. + */ + priceChf: number | null; + /** id of the underlying log row — logId is ALWAYS present, on every row, even the null-price ones. + * Overflow detection / keyset cursor logic in LedgerMarkService counts distinct logId values across + * ALL returned rows (not just the ones with a usable price) to know how many log rows were actually + * read — see LedgerMarkService.uniqueLogIds. */ + logId: number; +} + @Injectable() export class LogRepository extends BaseRepository { constructor(manager: EntityManager) { @@ -217,6 +238,92 @@ export class LogRepository extends BaseRepository { return rows; } + /** + * SQL-side projection of priceChf per asset from FinancialDataLog snapshots. + * LIMIT/keyset apply to log rows (inner subquery), not to the expanded asset result — same page semantics as + * getFinancialLogs. Callers that only need marks avoid shipping/parsing the full message JSON. + * + * LEFT JOIN LATERAL (not an implicit CROSS JOIN) so every log row yields at least one result row — including + * when `assets` is empty/absent or a key/price is unusable (assetId/priceChf null). That keeps logId-based + * overflow detection and keyset pagination in LedgerMarkService correct (Finding 1+2: the old join+WHERE + * dropped unusable-price rows entirely, so uniqueLogIds under-counted and pagination/overflow stopped early). + * Invalid keys and non-number priceChf are nulled via CASE expressions rather than filtered in WHERE, so the + * row (and its logId) always remains. Malformed `message` JSON fails loud: `message::jsonb` aborts the whole + * query (intentional change vs the old JS path that try/caught per row). + */ + async getFinancialLogAssetPrices( + from?: Date, + to?: Date, + limit?: number, + after?: number, // id of the last LOG row of the previous page; same cursor semantics as getFinancialLogs + ): Promise { + const params: unknown[] = []; + let i = 1; + const conditions = [`system = $${i++}`, `subsystem = $${i++}`, `severity = $${i++}`, `valid = $${i++}`]; + params.push('LogService', FINANCIAL_DATA_LOG_SUBSYSTEM, LogSeverity.INFO, true); + + if (from) { + conditions.push(`created >= $${i++}`); + params.push(from); + } + if (to) { + conditions.push(`created <= $${i++}`); + params.push(to); + } + if (after != null) { + // Same row-value keyset as getFinancialLogs: created resolved in-DB at full precision. + conditions.push(`(created, id) > ((SELECT c.created FROM log c WHERE c.id = $${i}), $${i + 1})`); + params.push(after, after); + i += 2; + } + + let limitClause = ''; + if (limit != null) { + limitClause = `LIMIT $${i++}`; + params.push(limit); + } + + const sql = ` +SELECT l.created AS "created", + CASE WHEN kv.key ~ '^[0-9]+$' THEN kv.key::int ELSE NULL END AS "assetId", + CASE + WHEN jsonb_typeof(kv.value -> 'priceChf') = 'number' THEN (kv.value ->> 'priceChf')::float8 + ELSE NULL + END AS "priceChf", + l.id AS "logId" +FROM ( + SELECT id, created, message + FROM log + WHERE ${conditions.join(' AND ')} + ORDER BY created ASC, id ASC + ${limitClause} +) l +LEFT JOIN LATERAL jsonb_each(l.message::jsonb -> 'assets') kv ON true +ORDER BY l.created ASC, l.id ASC`; + + const raw = (await this.query(sql, params)) as { + created: Date | string; + assetId: number | string | null; + priceChf: number | string | null; + logId: number | string; + }[]; + + const rows: FinancialLogAssetPrice[] = raw.map((r) => { + const priceChf = r.priceChf == null ? null : Number(r.priceChf); + return { + created: r.created instanceof Date ? r.created : new Date(r.created), + assetId: r.assetId == null ? null : Number(r.assetId), + // float8 NaN/Infinity (e.g. an out-of-range numeric text) must be excluded the same way the old + // Number.isFinite gate excluded them — never surface as a phantom mark. + priceChf: priceChf != null && Number.isFinite(priceChf) ? priceChf : null, + logId: Number(r.logId), + }; + }); + + if (!rows.length && after != null) await this.assertEmptyResultIsEndOfData(after); + return rows; + } + // After an empty main-query result with a keyset cursor, fail loud when the cursor id is gone: the row-value // subquery would return NULL and `(created, id) > (NULL, :afterId)` is NULL in Postgres → WHERE excludes every // row → silent empty result that callers misread as end-of-data. Only invoked when the main query already diff --git a/src/subdomains/supporting/log/log.service.ts b/src/subdomains/supporting/log/log.service.ts index edc180269a..d122a4708e 100644 --- a/src/subdomains/supporting/log/log.service.ts +++ b/src/subdomains/supporting/log/log.service.ts @@ -13,7 +13,7 @@ import { LogSeverity, MAX_VALIDITY_SWEEP_ROWS, } from './log.entity'; -import { LogRepository } from './log.repository'; +import { FinancialLogAssetPrice, LogRepository } from './log.repository'; @Injectable() export class LogService { @@ -145,6 +145,15 @@ export class LogService { return this.logRepo.getFinancialLogs(from, dailySample, to, limit, after); } + async getFinancialLogAssetPrices( + from?: Date, + to?: Date, + limit?: number, + after?: number, + ): Promise { + return this.logRepo.getFinancialLogAssetPrices(from, to, limit, after); + } + async getLatestFinancialLog(): Promise { return this.logRepo.getLatestFinancialLog(); }