From c8c1ec3ba21e230a25b293c2124c7bb6371982ec Mon Sep 17 00:00:00 2001 From: TaprootFreak <142087526+TaprootFreak@users.noreply.github.com> Date: Wed, 29 Jul 2026 17:41:02 +0200 Subject: [PATCH 1/2] perf(ledger): project asset prices in SQL instead of loading whole snapshots MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit LedgerMarkService.preload loaded complete log rows and parsed each message in Node, only to keep three values per asset: created, assetId and priceChf. A snapshot carries 200 assets in an average 42 KB JSON document. Measured on the real two-day preload window in production: 930 rows and 38 MB transferred, of which 7.3 MB is the data actually used. The JSON.parse over those 38 MB runs on the event loop and is a direct cause of the API container sitting at a saturated core, which slows every request. The new repository method projects the prices in SQL via LATERAL jsonb_each and returns flat rows. The LIMIT and the keyset cursor stay on the log rows, not on the expanded result, so pagination semantics are unchanged — the cursor therefore carries the originating log id. getLatestMarks keeps using getFinancialLogs: it runs on the dailySample path, where the row count is already small and the added complexity would not pay for itself. Mark map contents, ordering and the handling of missing or non-finite prices are unchanged. --- .../__tests__/ledger-mark.service.spec.ts | 236 ++++++++++++++---- .../services/ledger-mark.service.ts | 120 ++++++--- .../supporting/log/log.repository.ts | 86 +++++++ src/subdomains/supporting/log/log.service.ts | 11 +- 4 files changed, 377 insertions(+), 76 deletions(-) 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..31e0dd5365 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,49 @@ 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 filters as the repository path). */ +function logToAssetPrices(row: Log): FinancialLogAssetPrice[] { + try { + const assets = (JSON.parse(row.message) as { assets?: Record }).assets; + if (!assets) return []; + + const out: FinancialLogAssetPrice[] = []; + for (const [assetIdKey, assetLog] of Object.entries(assets)) { + const priceChf = assetLog?.priceChf; + if (!Number.isFinite(priceChf)) continue; + out.push({ + created: row.created, + assetId: +assetIdKey, + priceChf, + logId: row.id, + }); + } + return out; + } catch { + return []; + } +} + +/** + * 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 +186,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 +259,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 +276,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 +288,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,8 +300,10 @@ 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')); @@ -251,9 +311,12 @@ describe('LedgerMarkService', () => { }); it('never throws on malformed message JSON (defensive parse)', async () => { + // Hot path no longer parses message in the service; projection fake yields no prices for unparseable rows. 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' })]), + ); const cache = await service.preload(new Date('2026-06-01'), new Date('2026-06-01')); @@ -272,39 +335,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); + }); + + // 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 normally drop non-finite rows; inject them here so buildMarkMap's guard is covered too. + jest.spyOn(logService, 'getFinancialLogAssetPrices').mockResolvedValue([ + { created: t, assetId: 5, priceChf: 42, logId: 1 }, + { created: t, assetId: 6, priceChf: NaN, logId: 1 }, + { created: t, assetId: 7, priceChf: Infinity, 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 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. + // §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 +439,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 +458,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 +486,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 +503,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 +513,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 +530,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 +577,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 +588,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 +625,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 +640,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 +652,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); @@ -522,7 +669,8 @@ describe('LedgerMarkService', () => { }); // 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..e667777af7 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,110 @@ 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). + private rowToAssetPrices(row: Log): FinancialLogAssetPrice[] { + const assets = this.parseAssets(row.message); + if (!assets) return []; + + const result: FinancialLogAssetPrice[] = []; + for (const [assetIdKey, assetLog] of Object.entries(assets)) { + const priceChf = assetLog?.priceChf; + if (!Number.isFinite(priceChf)) continue; + + result.push({ + created: row.created, + assetId: +assetIdKey, + priceChf, + logId: row.id, + }); + } + return result; + } + + // Distinct logIds in first-seen order (SQL keeps all assets of one log contiguous). + 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 +255,19 @@ 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; + // Repo already drops non-finite values; keep the guard as a defensive double-check. + if (!Number.isFinite(row.priceChf)) continue; - const assetId = +assetIdKey; - const points = marks.get(assetId) ?? []; - points.push({ created: row.created, priceChf }); - marks.set(assetId, points); - } + 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..9cef78b709 100644 --- a/src/subdomains/supporting/log/log.repository.ts +++ b/src/subdomains/supporting/log/log.repository.ts @@ -19,6 +19,15 @@ 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; + assetId: number; + priceChf: number; + /** id of the underlying log row — keyset cursor / overflow counts use this, not result-row index. */ + logId: number; +} + @Injectable() export class LogRepository extends BaseRepository { constructor(manager: EntityManager) { @@ -217,6 +226,83 @@ 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. + */ + 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", + (kv.key)::int AS "assetId", + (kv.value->>'priceChf')::float8 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, +LATERAL jsonb_each(l.message::jsonb -> 'assets') kv +WHERE kv.value->>'priceChf' IS NOT NULL +ORDER BY l.created ASC, l.id ASC`; + + const raw = (await this.query(sql, params)) as { + created: Date | string; + assetId: number | string; + priceChf: number | string; + logId: number | string; + }[]; + + const rows: FinancialLogAssetPrice[] = []; + for (const r of raw) { + const priceChf = Number(r.priceChf); + if (!Number.isFinite(priceChf)) continue; + + rows.push({ + created: r.created instanceof Date ? r.created : new Date(r.created), + assetId: Number(r.assetId), + priceChf, + 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(); } From 4c324699117f4675088172823efa21e539056f8e Mon Sep 17 00:00:00 2001 From: TaprootFreak <142087526+TaprootFreak@users.noreply.github.com> Date: Wed, 29 Jul 2026 18:12:50 +0200 Subject: [PATCH 2/2] fix(ledger): keep every read log row countable and tighten price validation MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Review follow-ups on the SQL price projection. A log row whose assets yield no usable price used to vanish from the projection entirely. Overflow detection and keyset pagination count log rows, so a vanished row made them undercount: preload could miss that more data exists, and paginate could stop early — a silent loss of marks. The projection now uses LEFT JOIN LATERAL, so every read log row comes back even when it carries no usable price, and the counting is correct again. This also removes a related false end-of-data path in the stale-cursor guard. Price conversion now matches the previous Number.isFinite semantics: a price is only taken when the JSON value really is a number, so a string "1.25" is skipped as before rather than silently accepted by the float cast. Non-numeric asset keys no longer abort the whole query; they yield null and are skipped. Malformed message JSON now fails loudly instead of being swallowed — deliberate, and documented on the method. buildMarkMap keeps Number.isFinite as a second line of defence. The repository already nulls NaN and Infinity, but a NaN mark would corrupt a valuation silently, so that must not rest on a single guard. The test fake mirrors the SQL semantics instead of masking it, and three tests cover the cases: a log row without usable prices between two valid ones, a string price, and a non-numeric asset key. --- .../__tests__/ledger-mark.service.spec.ts | 106 ++++++++++++++---- .../services/ledger-mark.service.ts | 36 +++--- .../supporting/log/log.repository.ts | 61 ++++++---- 3 files changed, 146 insertions(+), 57 deletions(-) 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 31e0dd5365..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 @@ -69,27 +69,28 @@ type FakeGetFinancialLogAssetPrices = ( after?: number, ) => Promise; -/** Expand a Log fixture into the SQL projection shape (same filters as the repository path). */ +/** Expand a Log fixture into the SQL projection shape (same LEFT JOIN LATERAL semantics as the repository). */ function logToAssetPrices(row: Log): FinancialLogAssetPrice[] { - try { - const assets = (JSON.parse(row.message) as { assets?: Record }).assets; - if (!assets) return []; - - const out: FinancialLogAssetPrice[] = []; - for (const [assetIdKey, assetLog] of Object.entries(assets)) { - const priceChf = assetLog?.priceChf; - if (!Number.isFinite(priceChf)) continue; - out.push({ - created: row.created, - assetId: +assetIdKey, - priceChf, - logId: row.id, - }); - } - return out; - } catch { - return []; + // 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, + }; + }); } /** @@ -310,19 +311,41 @@ describe('LedgerMarkService', () => { expect(cache.getMarkAt(5, new Date('2026-06-01'))).toBeUndefined(); }); - it('never throws on malformed message JSON (defensive parse)', async () => { - // Hot path no longer parses message in the service; projection fake yields no prices for unparseable rows. + it('throws on malformed message JSON (fail-loud, matches `message::jsonb` in production SQL)', async () => { jest .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') @@ -398,11 +421,11 @@ describe('LedgerMarkService', () => { 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 normally drop non-finite rows; inject them here so buildMarkMap's guard is covered too. + // 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: NaN, logId: 1 }, - { created: t, assetId: 7, priceChf: Infinity, logId: 1 }, + { created: t, assetId: 6, priceChf: null, logId: 1 }, + { created: t, assetId: null, priceChf: 7, logId: 1 }, // asset 8 absent entirely → no mark ]); @@ -668,6 +691,41 @@ 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). Stays on getFinancialLogs (no dailySample // parameter on getFinancialLogAssetPrices) — rare long-window path only. diff --git a/src/subdomains/core/accounting/services/ledger-mark.service.ts b/src/subdomains/core/accounting/services/ledger-mark.service.ts index e667777af7..fda1173e47 100644 --- a/src/subdomains/core/accounting/services/ledger-mark.service.ts +++ b/src/subdomains/core/accounting/services/ledger-mark.service.ts @@ -208,26 +208,33 @@ export class LedgerMarkService { } // 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); - if (!assets) return []; + const entries = assets ? Object.entries(assets) : []; - const result: FinancialLogAssetPrice[] = []; - for (const [assetIdKey, assetLog] of Object.entries(assets)) { - const priceChf = assetLog?.priceChf; - if (!Number.isFinite(priceChf)) continue; + if (!entries.length) { + return [{ created: row.created, assetId: null, priceChf: null, logId: row.id }]; + } - result.push({ + return entries.map(([assetIdKey, assetLog]) => { + const priceChf = assetLog?.priceChf; + return { created: row.created, - assetId: +assetIdKey, - priceChf, + assetId: /^[0-9]+$/.test(assetIdKey) ? Number(assetIdKey) : null, + priceChf: typeof priceChf === 'number' && Number.isFinite(priceChf) ? priceChf : null, logId: row.id, - }); - } - return result; + }; + }); } // 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(); @@ -259,8 +266,11 @@ export class LedgerMarkService { const marks = new Map(); for (const row of rows) { - // Repo already drops non-finite values; keep the guard as a defensive double-check. - if (!Number.isFinite(row.priceChf)) continue; + // 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 }); diff --git a/src/subdomains/supporting/log/log.repository.ts b/src/subdomains/supporting/log/log.repository.ts index 9cef78b709..6be2809ef7 100644 --- a/src/subdomains/supporting/log/log.repository.ts +++ b/src/subdomains/supporting/log/log.repository.ts @@ -22,9 +22,21 @@ import { /** One asset priceChf projected from a FinancialDataLog snapshot (no full message JSON). */ export interface FinancialLogAssetPrice { created: Date; - assetId: number; - priceChf: number; - /** id of the underlying log row — keyset cursor / overflow counts use this, not result-row index. */ + /** + * 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; } @@ -230,6 +242,14 @@ export class LogRepository extends BaseRepository { * 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, @@ -265,8 +285,11 @@ export class LogRepository extends BaseRepository { const sql = ` SELECT l.created AS "created", - (kv.key)::int AS "assetId", - (kv.value->>'priceChf')::float8 AS "priceChf", + 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 @@ -274,30 +297,28 @@ FROM ( WHERE ${conditions.join(' AND ')} ORDER BY created ASC, id ASC ${limitClause} -) l, -LATERAL jsonb_each(l.message::jsonb -> 'assets') kv -WHERE kv.value->>'priceChf' IS NOT NULL +) 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; - priceChf: number | string; + assetId: number | string | null; + priceChf: number | string | null; logId: number | string; }[]; - const rows: FinancialLogAssetPrice[] = []; - for (const r of raw) { - const priceChf = Number(r.priceChf); - if (!Number.isFinite(priceChf)) continue; - - rows.push({ + 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: Number(r.assetId), - priceChf, + 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;