From 6b5bc5d7a556fbf650f13265a158e4da48883836 Mon Sep 17 00:00:00 2001 From: eleith Date: Tue, 10 Nov 2026 10:34:07 -0800 Subject: [PATCH] feat(snapshots): publish SQLite report generations --- .../publish-sqlite-report-generations.md | 6 + .../lib/server/reports.integration.test.ts | 155 +- .../cash/[accountId]/page.server.test.ts | 26 + .../[accountId]/page.server.test.ts | 19 + .../prepare-reports/drain.integration.test.ts | 106 -- .../prepare-reports/edits.integration.test.ts | 162 -- .../prepare-reports/run.integration.test.ts | 330 +---- .../src/workflows/prepare-reports/run.test.ts | 16 +- .../prepare-reports/test-fixtures.ts | 18 - docs/architecture.md | 7 +- docs/development.md | 1 + packages/db/src/database.ts | 5 +- packages/db/src/index.ts | 1 + packages/db/src/reports/queries.test.ts | 83 +- packages/snapshots/README.md | 4 +- .../snapshots/src/reports/publication.test.ts | 1319 ++++++----------- packages/snapshots/src/reports/publication.ts | 184 ++- 17 files changed, 828 insertions(+), 1614 deletions(-) create mode 100644 .changeset/publish-sqlite-report-generations.md delete mode 100644 apps/worker/src/workflows/prepare-reports/drain.integration.test.ts delete mode 100644 apps/worker/src/workflows/prepare-reports/edits.integration.test.ts diff --git a/.changeset/publish-sqlite-report-generations.md b/.changeset/publish-sqlite-report-generations.md new file mode 100644 index 0000000..6afdbc3 --- /dev/null +++ b/.changeset/publish-sqlite-report-generations.md @@ -0,0 +1,6 @@ +--- +'@hledger-reports/web': patch +'@hledger-reports/worker': patch +--- + +Store prepared financial reports in one private SQLite database per generation, keeping exact values and page behavior unchanged. After updating web and worker together, prepare reports again. Older financial JSON files are preserved, not migrated or deleted. diff --git a/apps/web/src/lib/server/reports.integration.test.ts b/apps/web/src/lib/server/reports.integration.test.ts index bd33b77..fd914e9 100644 --- a/apps/web/src/lib/server/reports.integration.test.ts +++ b/apps/web/src/lib/server/reports.integration.test.ts @@ -2,8 +2,13 @@ import { mkdtemp, rm } from 'node:fs/promises'; import { tmpdir } from 'node:os'; import { join } from 'node:path'; import { expect, it } from 'vitest'; -import type { OverviewData, PreparedAccountSummary } from '@hledger-reports/hledger'; -import { createReportStage } from '@hledger-reports/snapshots'; +import type { PreparedAccountRow, PreparedAccountSummary } from '@hledger-reports/hledger'; +import { createReportStage, readReportPublication } from '@hledger-reports/snapshots'; +import { + amount, + preparedAccountRows, + preparedOverview +} from '../../../../../packages/db/src/reports/data.fixture.js'; import { load as loadLayout } from '../../routes/(protected)/+layout.server.js'; import { load as loadCashAccount } from '../../routes/(protected)/registers/cash/[accountId]/+page.server.js'; @@ -17,44 +22,85 @@ it('keeps the layout and saved account leaf on one real publication until a fres journal: '/PRIVATE/fictional.journal', accounts: [{ account, label: 'Cash', register: 'cash' as const }] }, - paths: { snapshotDirectory } + paths: { snapshotDirectory }, + get jobsClient(): never { + throw new Error('Browsing must not access the queue.'); + } }; - // An undated, empty journal is a complete current-format report, not a partial fixture. - const overview: OverviewData = { - asOf: null, - preparedAt: '2026-05-01T12:00:00Z', - journalCoverage: { transactionCount: 0, firstDate: null, lastDate: null }, - hledgerVersion: '1.99.5', - ranges: null, - balanceSheet: null, - incomeExpenses: { months: [] }, - netWorth: { currency: null, points: [] } + const overview = { ...preparedOverview(), preparedAt: '2026-05-01T12:00:00Z' }; + const activity = (payee: string, balance: string) => { + const rows = preparedAccountRows(); + const row = rows[0]; + row.isSplit = false; + row.balance = [amount(balance)]; + row.transaction.payee = payee; + row.transaction.postings = row.transaction.postings.slice(0, 2); + row.transaction.postings[1].amounts = [amount('100.00')]; + row.transaction.postings[1].transactedAmounts = [amount('100.00')]; + row.investment.expensePostingIndexes = [1]; + return rows; }; - const accounts: PreparedAccountSummary[] = [ - { name: account, access: 'ready', rowCount: 0, summary: { status: 'undated' } } + const rowsA = activity('Market A', '9007199254740993.123456789'); + const rowsB = activity('Market B', '9007199254740994.123456789'); + const summary = (row: PreparedAccountRow): PreparedAccountSummary[] => [ + { + name: account, + access: 'ready', + rowCount: 1, + summary: { + status: 'ready', + balances: row.balance, + valuation: { amounts: row.balance, unconverted: [] } + } + } ]; const stageA = await createReportStage(snapshotDirectory, { id: 'A', - createdAt: new Date('2026-05-01T12:00:00Z') + createdAt: new Date(overview.preparedAt) }); - await stageA.publish(overview, accounts, application.config.journal); + await stageA.writeRows(account, rowsA); + await stageA.publish(overview, summary(rowsA[0]), application.config.journal); const locals = { application }; const layoutA = await loadLayout({ locals, url: new URL('http://localhost/registers/cash') } as unknown as Parameters[0]); - expect(layoutA.accountSidebar.hasSavedReports).toBe(true); const idA = layoutA.accountSidebar.groups[0].accounts[0].id!; + const publicBalanceA = { commodity: '$', quantity: '9007199254740993.123456789' }; + const publicBalanceB = { commodity: '$', quantity: '9007199254740994.123456789' }; + const expectedSidebarAccount = { + id: idA, + name: account, + label: 'Cash', + access: 'ready', + value: { status: 'ready', amounts: [publicBalanceA], unconverted: [] } + }; + const expectedLayoutA = { + accountLabels: [{ account, label: 'Cash' }], + pathname: '/registers/cash', + accountSidebar: { + asOf: overview.asOf, + hasSavedReports: true, + netWorth: { amounts: [publicBalanceA], unconverted: [] }, + readFailed: false, + groups: [{ kind: 'cash', accounts: [expectedSidebarAccount] }] + } + }; + expect(layoutA).toEqual(expectedLayoutA); + const previous = await readReportPublication(snapshotDirectory); + if (!previous) throw new Error('Expected publication A.'); const preparedB = '2026-05-02T12:00:00Z'; - const stageB = await createReportStage(snapshotDirectory, { - id: 'B', - createdAt: new Date(preparedB) - }); + const stageB = await createReportStage( + snapshotDirectory, + { id: 'B', createdAt: new Date(preparedB) }, + previous.accounts + ); + await stageB.writeRows(account, rowsB); await stageB.publish( { ...overview, preparedAt: preparedB }, - accounts, + summary(rowsB[0]), application.config.journal ); @@ -66,14 +112,35 @@ it('keeps the layout and saved account leaf on one real publication until a fres parent: async () => layoutA, setHeaders: () => {} } as unknown as Parameters[0]); - expect(leafA).toMatchObject({ + const publicChange = { commodity: '$', quantity: '-100.00' }; + const expectedRowA = { + index: 1, + date: '2026-04-30', + payee: 'Market A', + categories: ['expenses:食料'], + notes: ['supplies', 'transaction note\nproject: home\n', 'bank note'], + amount: publicChange, + balance: publicBalanceA, + statuses: ['Cleared'], + direction: 'payment', + breakdown: null + }; + const expectedLeafA = { + configured: true, preparedAt: overview.preparedAt, + accountId: idA, accountName: account, access: 'ready', - rowCount: 0, - rows: [], + rowCount: 1, + page: 1, + pageCount: 1, + order: 'asc', + rows: [expectedRowA], + endingBalance: publicBalanceA, + unsupported: false, readFailed: false - }); + }; + expect(leafA).toEqual(expectedLeafA); const freshLocals = { application }; const layoutB = await loadLayout({ @@ -81,6 +148,24 @@ it('keeps the layout and saved account leaf on one real publication until a fres url: new URL('http://localhost/registers/cash') } as unknown as Parameters[0]); const idB = layoutB.accountSidebar.groups[0].accounts[0].id!; + expect(idB).toBe(idA); + expect(layoutB).toEqual({ + ...expectedLayoutA, + accountSidebar: { + ...expectedLayoutA.accountSidebar, + groups: [ + { + kind: 'cash', + accounts: [ + { + ...expectedSidebarAccount, + value: { status: 'ready', amounts: [publicBalanceB], unconverted: [] } + } + ] + } + ] + } + }); const leafB = await loadCashAccount({ locals: freshLocals, url: new URL(`http://localhost/registers/cash/${idB}`), @@ -88,10 +173,20 @@ it('keeps the layout and saved account leaf on one real publication until a fres parent: async () => layoutB, setHeaders: () => {} } as unknown as Parameters[0]); - expect(layoutB.accountSidebar.hasSavedReports).toBe(true); - expect(leafB.preparedAt).toBe(preparedB); + expect(leafB).toEqual({ + ...expectedLeafA, + preparedAt: preparedB, + rows: [ + { + ...expectedRowA, + payee: 'Market B', + balance: publicBalanceB + } + ], + endingBalance: publicBalanceB + }); expect(JSON.stringify([layoutA, leafA, layoutB, leafB])).not.toMatch( - /PRIVATE|directory|journal|publication|savedReports/ + /PRIVATE|directory|journal|publication|savedReports|sourcePosition|costBasis|style/ ); } finally { await rm(root, { recursive: true, force: true }); diff --git a/apps/web/src/routes/(protected)/registers/cash/[accountId]/page.server.test.ts b/apps/web/src/routes/(protected)/registers/cash/[accountId]/page.server.test.ts index 989da39..52f09f6 100644 --- a/apps/web/src/routes/(protected)/registers/cash/[accountId]/page.server.test.ts +++ b/apps/web/src/routes/(protected)/registers/cash/[accountId]/page.server.test.ts @@ -105,6 +105,32 @@ it('loads a member from one pinned publication and delegates its requested page' expect(mapCashRegisterPage).toHaveBeenCalledExactlyOnceWith(rows, 'desc', 2); expect(request.setHeaders).toHaveBeenCalledWith({ 'cache-control': 'no-store' }); }); +it('returns a safe read failure when the saved body cannot be read', async () => { + vi.mocked(readAccountTransactions).mockRejectedValue( + new Error('PRIVATE /books/included.journal:12 database diagnostic') + ); + const request = event(); + const queue = vi.spyOn(request.locals.application!, 'jobsClient', 'get'); + const result = await load(request); + expect(result).toEqual({ + configured: true, + readFailed: true, + accountId: id, + accountName: name, + access: 'ready', + preparedAt: publication.overview.preparedAt, + rowCount: 51, + pageCount: 2, + page: 2, + order: 'asc', + rows: [], + endingBalance: null, + unsupported: false + }); + expect(readAccountTransactions).toHaveBeenCalledExactlyOnceWith(publication, id); + expect(mapCashRegisterPage).not.toHaveBeenCalled(); + expect(queue).not.toHaveBeenCalled(); +}); it('defaults ascending navigation to the last page', async () => { await load(event()); expect(mapCashRegisterPage).toHaveBeenCalledExactlyOnceWith(rows, 'asc', 2); diff --git a/apps/web/src/routes/(protected)/registers/investments/[accountId]/page.server.test.ts b/apps/web/src/routes/(protected)/registers/investments/[accountId]/page.server.test.ts index e6312c3..12e63ce 100644 --- a/apps/web/src/routes/(protected)/registers/investments/[accountId]/page.server.test.ts +++ b/apps/web/src/routes/(protected)/registers/investments/[accountId]/page.server.test.ts @@ -111,6 +111,25 @@ it('loads a member from one pinned publication and delegates its requested page' expect(mapInvestmentRegisterPage).toHaveBeenCalledExactlyOnceWith(rows, 'desc', 2); expect(request.setHeaders).toHaveBeenCalledWith({ 'cache-control': 'no-store' }); }); +it('returns a safe read failure when the saved body cannot be read', async () => { + vi.mocked(readAccountTransactions).mockRejectedValue( + new Error('PRIVATE /books/included.journal:12 database diagnostic') + ); + const request = event(); + const queue = vi.spyOn(request.locals.application!, 'jobsClient', 'get'); + const result = await load(request); + expect(result).toEqual({ + configured: true, + readFailed: true, + accountName: name, + access: 'ready', + preparedAt: publication.overview.preparedAt, + register: null + }); + expect(readAccountTransactions).toHaveBeenCalledExactlyOnceWith(publication, id); + expect(mapInvestmentRegisterPage).not.toHaveBeenCalled(); + expect(queue).not.toHaveBeenCalled(); +}); it('defaults ascending navigation to the last page', async () => { await load(event()); expect(mapInvestmentRegisterPage).toHaveBeenCalledExactlyOnceWith(rows, 'asc', 2); diff --git a/apps/worker/src/workflows/prepare-reports/drain.integration.test.ts b/apps/worker/src/workflows/prepare-reports/drain.integration.test.ts deleted file mode 100644 index 7f1aaa0..0000000 --- a/apps/worker/src/workflows/prepare-reports/drain.integration.test.ts +++ /dev/null @@ -1,106 +0,0 @@ -import { enqueuePreparation } from './test-fixtures.js'; -import { mkdir, mkdtemp, readFile, readdir, rm, writeFile } from 'node:fs/promises'; -import { tmpdir } from 'node:os'; -import { join } from 'node:path'; -import { afterEach, expect, it, vi } from 'vitest'; -import { prepareReports } from '@hledger-reports/hledger'; -import { openJobsClient, prepareReportsSpec } from '@hledger-reports/jobs'; -import { readOverview, readReportPublication } from '@hledger-reports/snapshots'; -import { lotsJournal } from '../../../fixtures/journal.js'; -import { startWorker } from '../../index.js'; - -vi.mock('@hledger-reports/hledger', async (original) => { - const actual = await original(); - return { ...actual, prepareReports: vi.fn(actual.prepareReports) }; -}); -const actual = await vi.importActual( - '@hledger-reports/hledger' -); -afterEach(() => { - vi.mocked(prepareReports).mockReset().mockImplementation(actual.prepareReports); -}); - -async function completeSets(snapshots: string) { - return (await readdir(join(snapshots, '.sets'))).filter((name) => !name.startsWith('.')); -} - -it('serves health and last-good data during deferred native preparation without a partial publication', async () => { - const directory = await mkdtemp(join(tmpdir(), 'reports-startup-')); - const config = join(directory, 'config', 'hledger-reports.yaml'); - const journal = join(directory, 'source.journal'); - const snapshots = join(directory, 'data', 'snapshots'); - await mkdir(join(directory, 'config')); - await writeFile( - config, - JSON.stringify({ - journal, - url: { app: 'http://127.0.0.1:3000' }, - auth: { credentials: { username: 'fixture', password: '${AUTH_PASSWORD}' } } - }) - ); - await writeFile(journal, lotsJournal); - const client = await openJobsClient(join(directory, 'data', 'jobs')); - let service: Awaited> | undefined; - let release!: () => void; - const gate = new Promise((resolve) => (release = resolve)); - const runs = async () => - (await client.backend.listWorkflowRuns({ workflowName: prepareReportsSpec.name })).data; - try { - service = await startWorker(config, { port: 0 }); - await enqueuePreparation(client); - await expect.poll(async () => (await runs())[0]?.status, { timeout: 12_000 }).toBe('completed'); - await service.stop(); - service = undefined; - const publication = (await readReportPublication(snapshots))!; - const original = publication.overview; - const sets = await completeSets(snapshots); - expect(sets).toHaveLength(1); - expect(original).not.toBeNull(); - expect( - JSON.parse(await readFile(join(snapshots, 'current', 'overview.json'), 'utf8')) - ).toMatchObject({ overview: original }); - // Edits alone do not prepare data; explicitly request the deferred run below. - await writeFile( - journal, - (await readFile(journal, 'utf8')).replace('Employer | Salary', 'Employer | Updated salary') - ); - let captured = false; - vi.mocked(prepareReports).mockImplementationOnce(async (...args) => { - const reports = await actual.prepareReports(...args); - captured = true; - await gate; - return reports; - }); - // Returning from start is not waiting for prepareReports or a job result. - service = await startWorker(config, { port: 0 }); - await enqueuePreparation(client); - await expect.poll(() => captured, { timeout: 12_000 }).toBe(true); - expect((await fetch(`http://127.0.0.1:${service.port}/healthz`)).status).toBe(200); - expect(await readReportPublication(snapshots)).toEqual(publication); - expect(await completeSets(snapshots)).toEqual(sets); - expect(await runs()).toEqual([ - expect.objectContaining({ status: 'running', output: null, attempts: 1 }), - expect.objectContaining({ status: 'completed', output: original }) - ]); - const stopping = service.stop(); - let stopped = false; - void stopping.then(() => (stopped = true)); - expect(stopped).toBe(false); - // Graceful stop keeps health/client alive until the active SDK handler finishes. - expect((await fetch(`http://127.0.0.1:${service.port}/healthz`)).status).toBe(200); - release(); - await stopping; - service = undefined; - const completed = await runs(); - expect(completed).toHaveLength(2); - expect(completed[0]).toMatchObject({ status: 'completed', attempts: 1 }); - expect(await readOverview(snapshots)).toEqual(completed[0]!.output); - expect(await completeSets(snapshots)).toHaveLength(2); - expect(prepareReports).toHaveBeenCalledTimes(2); - } finally { - release(); - await service?.stop(); - await client.close(); - await rm(directory, { recursive: true, force: true }); - } -}, 30_000); diff --git a/apps/worker/src/workflows/prepare-reports/edits.integration.test.ts b/apps/worker/src/workflows/prepare-reports/edits.integration.test.ts deleted file mode 100644 index 8d02111..0000000 --- a/apps/worker/src/workflows/prepare-reports/edits.integration.test.ts +++ /dev/null @@ -1,162 +0,0 @@ -import { enqueuePreparation, financialValues } from './test-fixtures.js'; -import { mkdir, mkdtemp, rm, writeFile } from 'node:fs/promises'; -import { tmpdir } from 'node:os'; -import { join } from 'node:path'; -import { setTimeout as delay } from 'node:timers/promises'; -import { afterEach, expect, it, vi } from 'vitest'; -import { prepareReports } from '@hledger-reports/hledger'; -import { openJobsClient, prepareReportsSpec, type JobsClient } from '@hledger-reports/jobs'; -import { - readReportPublication, - readAccountTransactions, - type ReportPublication -} from '@hledger-reports/snapshots'; -import { lotsJournal as original } from '../../../fixtures/journal.js'; -import { startWorker } from '../../index.js'; - -vi.mock('@hledger-reports/hledger', async (original) => { - const actual = await original(); - return { ...actual, prepareReports: vi.fn(actual.prepareReports) }; -}); -const hledger = await vi.importActual( - '@hledger-reports/hledger' -); -const directories: string[] = []; -afterEach(async () => { - vi.mocked(prepareReports).mockReset().mockImplementation(hledger.prepareReports); - await Promise.all( - directories.splice(0).map((path) => rm(path, { recursive: true, force: true })) - ); -}); -async function fixture() { - const directory = await mkdtemp(join(tmpdir(), 'reports-source-')); - directories.push(directory); - const config = join(directory, 'config', 'hledger-reports.yaml'); - const journal = join(directory, 'main.journal'); - const included = join(directory, 'parts', 'base.journal'); - const snapshots = join(directory, 'data', 'snapshots'); - await mkdir(join(directory, 'config')); - await mkdir(join(directory, 'parts')); - await writeFile( - config, - JSON.stringify({ - journal, - url: { app: 'http://127.0.0.1:3000' }, - auth: { credentials: { username: 'fixture', password: '${AUTH_PASSWORD}' } } - }) - ); - await writeFile(journal, 'include parts/*.journal\n'); - await writeFile(included, original); - return { directory, config, journal, included, snapshots, jobs: join(directory, 'data', 'jobs') }; -} -const runs = async (client: JobsClient) => - (await client.backend.listWorkflowRuns({ workflowName: prepareReportsSpec.name })).data; -async function settled(client: JobsClient, count: number) { - await expect - .poll( - async () => { - const all = await runs(client); - return { count: all.length, status: all[0]?.status }; - }, - { timeout: 15_000 } - ) - .toEqual({ count, status: 'completed' }); -} -function salary(value: string) { - return original - .replace('1000.00 USD\n income:salary', `${value} USD\n income:salary`) - .replace(/(income:salary\s+)-1000\.00 USD/, `$1-${value} USD`); -} -async function expectSalaryTransactions(publication: ReportPublication, quantity: string) { - expect(publication.directory).toBeTypeOf('string'); - const salary = publication.accounts?.find((account) => account.name === 'income:salary'); - expect(salary).toMatchObject({ access: 'ready', rowCount: 1 }); - const { transactions: rows } = await readAccountTransactions(publication, salary!.id); - expect(rows).toHaveLength(1); - expect(financialValues(rows![0].change)).toEqual([{ commodity: 'USD', quantity }]); -} -function gate() { - let release!: () => void; - let released = false; - const waiting = new Promise((resolve) => { - release = () => { - released = true; - resolve(); - }; - }); - return { - waiting, - release, - get released() { - return released; - } - }; -} - -it('publishes its loaded native view despite later edits, with no automatic follow-up', async () => { - const f = await fixture(); - const client = await openJobsClient(f.jobs); - const service = await startWorker(f.config, { port: 0 }); - const firstRead = gate(), - firstRelease = gate(), - nextRead = gate(), - nextRelease = gate(); - try { - await enqueuePreparation(client); - await settled(client, 1); - const initial = (await readReportPublication(f.snapshots))!; - await expectSalaryTransactions(initial, '-1000.00'); - vi.mocked(prepareReports) - .mockImplementationOnce(async (...args) => { - const data = await hledger.prepareReports(...args); - firstRead.release(); - await firstRelease.waiting; - return data; - }) - .mockImplementationOnce(async (...args) => { - const data = await hledger.prepareReports(...args); - nextRead.release(); - await nextRelease.waiting; - return data; - }); - await writeFile(f.included, salary('1100.00')); - await enqueuePreparation(client); - await expect.poll(() => firstRead.released, { timeout: 12_000 }).toBe(true); - await writeFile(f.included, salary('1200.00')); - await writeFile(f.included, salary('1300.00')); - await delay(2200); - expect(await runs(client)).toHaveLength(2); - expect(await readReportPublication(f.snapshots)).toEqual(initial); - firstRelease.release(); - await settled(client, 2); - await delay(2200); - expect(await runs(client)).toHaveLength(2); - expect(nextRead.released).toBe(false); - const intervening = (await readReportPublication(f.snapshots))!; - expect(intervening.overview.incomeExpenses.months[11].income).toEqual([ - { commodity: 'USD', quantity: '1100.00' } - ]); - expect(intervening.directory).not.toBe(initial.directory); - await expectSalaryTransactions(intervening, '-1100.00'); - await enqueuePreparation(client); - await expect.poll(() => nextRead.released, { timeout: 12_000 }).toBe(true); - expect(await runs(client)).toHaveLength(3); - nextRelease.release(); - await settled(client, 3); - const latest = (await readReportPublication(f.snapshots))!; - expect(latest.overview.incomeExpenses.months[11].income).toEqual([ - { commodity: 'USD', quantity: '1300.00' } - ]); - expect(latest.directory).not.toBe(intervening.directory); - await expectSalaryTransactions(latest, '-1300.00'); - await expectSalaryTransactions(initial, '-1000.00'); - await delay(2200); - expect(await runs(client)).toHaveLength(3); - expect(prepareReports).toHaveBeenCalledTimes(3); - } finally { - firstRelease.release(); - nextRelease.release(); - await service.stop(); - await client.close(); - } -}, 40_000); diff --git a/apps/worker/src/workflows/prepare-reports/run.integration.test.ts b/apps/worker/src/workflows/prepare-reports/run.integration.test.ts index a10dff4..194bf68 100644 --- a/apps/worker/src/workflows/prepare-reports/run.integration.test.ts +++ b/apps/worker/src/workflows/prepare-reports/run.integration.test.ts @@ -1,17 +1,11 @@ -import { enqueuePreparation, readLatestPreparationState } from './test-fixtures.js'; +import { enqueuePreparation } from './test-fixtures.js'; import { mkdir, mkdtemp, readFile, readdir, rm, writeFile } from 'node:fs/promises'; import { tmpdir } from 'node:os'; -import { setTimeout as delay } from 'node:timers/promises'; import { join } from 'node:path'; import { afterEach, expect, it } from 'vitest'; +import { openJobsClient, prepareReportsSpec, type JobsClient } from '@hledger-reports/jobs'; import { - openJobsClient, - prepareReportsSpec, - liveBalanceQuerySpec, - type JobsClient, - retiredIncomePreparationSpec -} from '@hledger-reports/jobs'; -import { + createReportStage, readAccountTransactions, readOverview, readReportPublication, @@ -45,17 +39,14 @@ async function fixture() { ); await writeFile(journal, 'include included.journal\n'); await writeFile(included, source); - const legacy = join(snapshots, 'income-expenses', 'journal', 'old-preview.json'); - await mkdir(join(snapshots, 'income-expenses', 'journal'), { recursive: true }); - await writeFile(legacy, 'legacy preview unchanged'); - return { directory, config, journal, included, jobs, snapshots, legacy }; + return { directory, config, journal, included, jobs, snapshots }; } async function preparations(client: JobsClient) { return (await client.backend.listWorkflowRuns({ workflowName: prepareReportsSpec.name })).data; } -async function settled(client: JobsClient, count: number, status = 'completed') { +async function settled(client: JobsClient, status: 'completed' | 'failed' = 'completed') { await expect .poll( async () => { @@ -64,7 +55,7 @@ async function settled(client: JobsClient, count: number, status = 'completed') }, { timeout: 12_000 } ) - .toEqual({ count, status }); + .toEqual({ count: 1, status }); return (await preparations(client))[0]!; } @@ -76,7 +67,7 @@ it('explicitly prepares a saved overview through the real SDK without a browser expect(await readOverview(f.snapshots)).toBeNull(); service = await startWorker(f.config, { port: 0 }); await enqueuePreparation(client); - const run = await settled(client, 1); + const run = await settled(client); expect(run).toMatchObject({ workflowName: prepareReportsSpec.name, version: '1', @@ -122,298 +113,59 @@ it('explicitly prepares a saved overview through the real SDK without a browser ); expect(await readFile(f.journal, 'utf8')).toBe('include included.journal\n'); expect(await readFile(f.included, 'utf8')).toBe(source); - expect(await readFile(f.legacy, 'utf8')).toBe('legacy preview unchanged'); } finally { await service?.stop(); await client.close(); } }, 20_000); -it('retains last-good data without retry flooding, permits an explicit SDK retry, and prepares after an explicit correction request', async () => { +it('persists a sanitized failed SDK request and completed validation without replacing saved reports', async () => { const f = await fixture(); + const stage = await createReportStage(f.snapshots, { + id: 'saved', + createdAt: new Date('2026-05-01T00:00:00Z') + }); + try { + await stage.publish( + { + asOf: null, + preparedAt: '2026-05-01T00:00:00.000Z', + journalCoverage: { transactionCount: 0, firstDate: null, lastDate: null }, + hledgerVersion: '1.99.5', + ranges: null, + balanceSheet: null, + incomeExpenses: { months: [] }, + netWorth: { currency: null, points: [] } + }, + [], + f.journal + ); + } finally { + await stage.discard(); + } + const saved = await readReportPublication(f.snapshots); + expect(saved).not.toBeNull(); + const savedSets = await readdir(join(f.snapshots, '.sets')); + await writeFile(f.included, 'BROKEN PRIVATE JOURNAL 12345.67 USD'); const client = await openJobsClient(f.jobs); let service: Awaited> | undefined; try { service = await startWorker(f.config, { port: 0 }); await enqueuePreparation(client); - await settled(client, 1); - const first = await readOverview(f.snapshots); - const savedSets = await readdir(join(f.snapshots, '.sets')); - await service.stop(); - service = undefined; - await writeFile(f.included, 'BROKEN PRIVATE JOURNAL 12345.67 USD'); - service = await startWorker(f.config, { port: 0 }); - await enqueuePreparation(client); - const failed = await settled(client, 2, 'failed'); - expect(failed.attempts).toBe(1); - expect(await readLatestPreparationState(client)).toBe('failed'); - expect(await readdir(join(f.snapshots, '.sets'))).toEqual(savedSets); + const failed = await settled(client, 'failed'); expect(failed.output).toBeNull(); - expect(failed.error?.message).toBe( - 'hledger strict journal validation failed. Run hledger check --strict directly for diagnostics.' - ); + expect(failed.error?.message).toBeTruthy(); + expect(JSON.stringify(failed.error)).not.toMatch(/PRIVATE|12345\.67/); + expect(JSON.stringify(failed.error)).not.toContain(f.directory); expect(await readValidation(f.snapshots)).toMatchObject({ journal: f.journal, - request: { id: failed.id }, + request: { id: failed.id, createdAt: failed.createdAt.toISOString() }, result: { outcome: 'could-not-validate', completedAt: expect.any(String) } }); - expect(JSON.stringify(failed.error)).not.toContain('PRIVATE'); - expect(JSON.stringify(failed.error)).not.toContain(f.directory); - expect(await readOverview(f.snapshots)).toEqual(first); - expect((await fetch(`http://127.0.0.1:${service.port}/healthz`)).status).toBe(200); - await delay(2100); - expect(await preparations(client)).toHaveLength(2); - await service.stop(); - service = await startWorker(f.config, { port: 0 }); - await delay(2100); - // Restart never requests a financial job, even after failure. - expect(await preparations(client)).toHaveLength(2); - // An explicit retry is still allowed for the same invalid input state. - await enqueuePreparation(client); - await settled(client, 3, 'failed'); - expect(await readOverview(f.snapshots)).toEqual(first); - const failedValidation = await readValidation(f.snapshots); - expect(failedValidation).toMatchObject({ result: { outcome: 'could-not-validate' } }); - await writeFile(f.included, changedSalary()); - expect( - ( - await ( - await client.workflow.runWorkflow(liveBalanceQuerySpec, { - query: 'type:X', - asOf: '2026-04-30' - }) - ).result({ timeoutMs: 10_000 }) - ).total - ).toEqual([{ commodity: 'USD', quantity: '100.00' }]); - // Correction and a live query neither request preparation nor clear a failure. - await delay(1500); - expect(await readValidation(f.snapshots)).toEqual(failedValidation); - expect(await readLatestPreparationState(client)).toBe('failed'); - expect(await preparations(client)).toHaveLength(3); - await enqueuePreparation(client); - const retried = await settled(client, 4); - const updated = (await readOverview(f.snapshots))!; - expect(updated.incomeExpenses.months[11].income).toEqual([ - { commodity: 'USD', quantity: '1500.00' } - ]); - expect(retried.output).toEqual(updated); - expect(await readValidation(f.snapshots)).toMatchObject({ - request: { id: retried.id }, - result: { outcome: 'passed' } - }); - await writeFile(f.included, ''); - await enqueuePreparation(client); - const emptyRun = await settled(client, 5); - const empty = await readOverview(f.snapshots); - expect(empty).toMatchObject({ - asOf: null, - incomeExpenses: { months: [] }, - ranges: null, - balanceSheet: null, - journalCoverage: { transactionCount: 0, firstDate: null, lastDate: null }, - netWorth: { currency: null, points: [] } - }); - expect(emptyRun.output).toEqual(empty); - expect(await readFile(f.legacy, 'utf8')).toBe('legacy preview unchanged'); + expect(await readReportPublication(f.snapshots)).toEqual(saved); + expect(await readdir(join(f.snapshots, '.sets'))).toEqual(savedSets); } finally { await service?.stop(); await client.close(); } -}, 30_000); - -it('fails old queued preview requests once without interpreting their input and prepares reports only on request', async () => { - const f = await fixture(); - const client = await openJobsClient(f.jobs); - const handle = await client.workflow.runWorkflow(retiredIncomePreparationSpec, { - old: 'opaque preview' - }); - const service = await startWorker(f.config, { port: 0 }); - try { - await expect(handle.result({ timeoutMs: 12_000 })).rejects.toThrow(); - const retired = await client.backend.getWorkflowRun({ workflowRunId: handle.workflowRun.id }); - expect(retired?.attempts).toBe(1); - expect(retired?.error?.message).toBe( - 'Income preparation previews are retired. Refresh the dashboard.' - ); - expect(await preparations(client)).toEqual([]); - await enqueuePreparation(client); - const prepared = await settled(client, 1); - expect(prepared.output).toEqual(await readOverview(f.snapshots)); - expect(await readOverview(f.snapshots)).toMatchObject({ asOf: '2026-04-16' }); - expect(await readFile(f.legacy, 'utf8')).toBe('legacy preview unchanged'); - } finally { - await service.stop(); - await client.close(); - } }, 20_000); - -function changedSalary() { - return source - .replace('1000.00 USD\n income:salary', '1500.00 USD\n income:salary') - .replace(/(income:salary\s+)-1000\.00 USD/, '$1-1500.00 USD'); -} - -it('prepares on request after a stopped-worker include edit even with saved data and an unchanged as-of date', async () => { - const f = await fixture(); - const client = await openJobsClient(f.jobs); - let service: Awaited> | undefined; - try { - service = await startWorker(f.config, { port: 0 }); - await enqueuePreparation(client); - const originalRun = await settled(client, 1); - await service.stop(); - service = undefined; - const original = (await readOverview(f.snapshots))!; - await writeFile(f.included, changedSalary()); - // Last-good data is readable before the restart, with no freshness inference. - expect(await readOverview(f.snapshots)).toEqual(original); - service = await startWorker(f.config, { port: 0 }); - await delay(1500); - expect(await preparations(client)).toHaveLength(1); - expect(await readOverview(f.snapshots)).toEqual(original); - await enqueuePreparation(client); - const refreshedRun = await settled(client, 2); - const refreshed = (await readOverview(f.snapshots))!; - expect(refreshedRun.id).not.toBe(originalRun.id); - expect(refreshedRun.idempotencyKey).toBeNull(); - expect(refreshed.asOf).toBe(original.asOf); - expect(refreshed.journalCoverage).toEqual(original.journalCoverage); - expect(refreshed.incomeExpenses.months[11].income).toEqual([ - { commodity: 'USD', quantity: '1500.00' } - ]); - expect(refreshed.netWorth?.points[11].amounts).toEqual([ - { commodity: 'USD', quantity: '12325.00' } - ]); - expect(refreshedRun.output).toEqual(refreshed); - expect(refreshedRun.output).not.toEqual(originalRun.output); - expect(await readFile(f.journal, 'utf8')).toBe('include included.journal\n'); - expect(await readFile(f.included, 'utf8')).toBe(changedSalary()); - } finally { - await service?.stop(); - await client.close(); - } -}, 30_000); - -it.each(['pending', 'abandoned running'] as const)( - 'executes previously requested %s preparation on boot without creating another run', - async (state) => { - const f = await fixture(); - const client = await openJobsClient(f.jobs); - let service: Awaited> | undefined; - try { - service = await startWorker(f.config, { port: 0 }); - await enqueuePreparation(client); - await settled(client, 1); - await service.stop(); - service = undefined; - const original = await readOverview(f.snapshots); - // A previously enqueued run must execute without another boot submission. - const pending = await client.workflow.runWorkflow(prepareReportsSpec, undefined); - if (state === 'abandoned running') { - const claimed = await client.backend.claimWorkflowRun({ - workerId: 'dead-worker', - leaseDurationMs: 100 - }); - expect(claimed).toMatchObject({ - id: pending.workflowRun.id, - status: 'running', - attempts: 1 - }); - const step = await client.backend.createStepAttempt({ - workflowRunId: pending.workflowRun.id, - workerId: 'dead-worker', - stepName: 'input-fingerprint', - stepIndex: 0, - kind: 'function', - config: {}, - context: null - }); - await client.backend.completeStepAttempt({ - workflowRunId: pending.workflowRun.id, - stepAttemptId: step.id, - workerId: 'dead-worker', - output: 'legacy-input-fingerprint' - }); - await expect.poll(() => Date.now() > claimed!.availableAt!.getTime()).toBe(true); - } - // An old checkpoint is ignored: recovery calculates the currently loaded journal. - await writeFile(f.included, changedSalary()); - expect(await readOverview(f.snapshots)).toEqual(original); - service = await startWorker(f.config, { port: 0 }); - const recovered = await settled(client, 2); - expect(recovered.id).toBe(pending.workflowRun.id); - // SDK 0.11 reclaims expired leases and executes the handler again, even - // with maximumAttempts=1 (that policy limits failure retries, not claims). - expect(recovered.attempts).toBe(state === 'pending' ? 1 : 2); - expect(recovered.output).toEqual(await readOverview(f.snapshots)); - const publication = (await readReportPublication(f.snapshots))!; - expect(publication.journal).toBe(f.journal); - if (state === 'abandoned running') { - const persisted = await client.backend.listStepAttempts({ workflowRunId: recovered.id }); - expect(persisted.data).toHaveLength(1); - expect(persisted.data[0]?.output).toBe('legacy-input-fingerprint'); - } - await delay(1500); - expect(await preparations(client)).toHaveLength(2); - expect(await readOverview(f.snapshots)).toMatchObject({ - incomeExpenses: { - months: expect.arrayContaining([ - expect.objectContaining({ income: [{ commodity: 'USD', quantity: '1500.00' }] }) - ]) - }, - netWorth: { - currency: 'USD', - points: expect.arrayContaining([ - expect.objectContaining({ - date: '2026-04-16', - amounts: [{ commodity: 'USD', quantity: '12325.00' }] - }) - ]) - } - }); - expect(await readFile(f.legacy, 'utf8')).toBe('legacy preview unchanged'); - } finally { - await service?.stop(); - await client.close(); - } - }, - 30_000 -); - -it('creates no financial work on boot, idle, edits or restart; every completed rerun needs an explicit request', async () => { - const f = await fixture(); - const client = await openJobsClient(f.jobs); - let service: Awaited> | undefined; - try { - service = await startWorker(f.config, { port: 0 }); - await delay(1500); - expect(await preparations(client)).toEqual([]); - expect(await readLatestPreparationState(client)).toBe('idle'); - expect(await readOverview(f.snapshots)).toBeNull(); - expect(await readValidation(f.snapshots)).toBeNull(); - await writeFile(f.included, changedSalary()); - await delay(1500); - expect(await preparations(client)).toEqual([]); - await service.stop(); - service = await startWorker(f.config, { port: 0 }); - await delay(1500); - expect(await preparations(client)).toEqual([]); - await enqueuePreparation(client); - const first = await settled(client, 1); - await delay(1500); - expect(await preparations(client)).toHaveLength(1); - await enqueuePreparation(client); - const second = await settled(client, 2); - expect(second.id).not.toBe(first.id); - expect(second.idempotencyKey).toBeNull(); - expect(await readLatestPreparationState(client)).toBe('idle'); - expect(second.output).toEqual(await readOverview(f.snapshots)); - await service.stop(); - service = await startWorker(f.config, { port: 0 }); - await delay(1500); - expect(await preparations(client)).toHaveLength(2); - } finally { - await service?.stop(); - await client.close(); - } -}, 30_000); diff --git a/apps/worker/src/workflows/prepare-reports/run.test.ts b/apps/worker/src/workflows/prepare-reports/run.test.ts index 1921945..30e39d1 100644 --- a/apps/worker/src/workflows/prepare-reports/run.test.ts +++ b/apps/worker/src/workflows/prepare-reports/run.test.ts @@ -125,7 +125,14 @@ it.each([true, false])( ...previous, journal: same ? journal : '/other/book.journal' }); - expect(await prepare(context)).toBe(overview); + const result = await prepare(context); + expect(createReportStage).toHaveBeenCalledExactlyOnceWith( + directory, + run, + same ? previous.accounts : null + ); + if (!same) return; + expect(result).toBe(overview); expect(checkJournal).toHaveBeenCalledExactlyOnceWith(journal); expect(writeValidation).toHaveBeenCalledExactlyOnceWith(directory, run, journal, { outcome: 'passed', @@ -134,11 +141,6 @@ it.each([true, false])( expect(vi.mocked(writeValidation).mock.invocationCallOrder[0]).toBeLessThan( vi.mocked(createReportStage).mock.invocationCallOrder[0]! ); - expect(createReportStage).toHaveBeenCalledExactlyOnceWith( - directory, - run, - same ? previous.accounts : null - ); expect(prepareReports).toHaveBeenCalledExactlyOnceWith(journal, { writeRows: stage.writeRows }); @@ -179,7 +181,7 @@ it.each(['producer', 'publish'] as const)('discards its stage on %s failure', as if (kind !== 'publish') expect(stage.publish).not.toHaveBeenCalled(); }); -it('awaits producer writes and publication before returning its own calculated result', async () => { +it('awaits producer completion and publication before returning its own calculated result', async () => { let releaseWrite!: () => void; let releasePublish!: () => void; stage.writeRows.mockReturnValue(new Promise((resolve) => (releaseWrite = resolve))); diff --git a/apps/worker/src/workflows/prepare-reports/test-fixtures.ts b/apps/worker/src/workflows/prepare-reports/test-fixtures.ts index 68aa61c..b287db1 100644 --- a/apps/worker/src/workflows/prepare-reports/test-fixtures.ts +++ b/apps/worker/src/workflows/prepare-reports/test-fixtures.ts @@ -1,24 +1,6 @@ -import type { ReportedAmount } from '@hledger-reports/hledger'; import { prepareReportsSpec, type JobsClient } from '@hledger-reports/jobs'; -/** Compare every exact financial value independently of retained display/cost metadata. */ -export function financialValues(amounts: readonly ReportedAmount[]) { - return amounts.map(({ commodity, quantity }) => ({ commodity, quantity })); -} - /** Direct SDK submission only; acceptance does not wait for a gated handler. */ export async function enqueuePreparation(client: JobsClient): Promise { await client.workflow.runWorkflow(prepareReportsSpec, undefined); } - -/** These isolated fixtures have one worker and explicitly sequenced preparations. */ -export async function readLatestPreparationState(client: JobsClient) { - const runs = await client.backend.listWorkflowRuns({ - workflowName: prepareReportsSpec.name, - limit: 1 - }); - const latest = runs.data[0]; - if (!latest || latest.status === 'completed') return 'idle'; - if (latest.status === 'failed' || latest.status === 'canceled') return 'failed'; - return latest.status; -} diff --git a/docs/architecture.md b/docs/architecture.md index 45c896c..88cae4f 100644 --- a/docs/architecture.md +++ b/docs/architecture.md @@ -17,9 +17,10 @@ A request keeps one saved set so navigation and page data agree. Recorded journa checks are separate from financial publication. Neither saved results nor a passed check proves that the journal has not changed. -Web and worker share configuration and private storage. SQLite coordinates jobs; -financial browsing does not depend on a running worker. Startup does not request -preparation, though previously requested jobs can resume. +Web and worker share configuration and private storage. The job queue and saved +report generations use separate SQLite databases. Financial browsing does not +depend on a running worker. Startup does not request preparation, though previously +requested jobs can resume. See [development](development.md) for commands and [principles](philosophy.md) for how we change the code. diff --git a/docs/development.md b/docs/development.md index 53c07a1..ec7c002 100644 --- a/docs/development.md +++ b/docs/development.md @@ -60,6 +60,7 @@ cleans up fictional data; never point browser tests at a personal deployment. - Keep live configuration, journals and runtime data out of Git and images. - Compose shares `apps/web/data/` between web and worker. It contains sensitive jobs and financial results and must be writable by container UID 1000. - Stop both services before backing up the whole data directory. Upgrade web and worker together; restore a consistent backup rather than downgrade a migrated database. +- Saved reports use SQLite generations. After updating web and worker together, explicitly prepare again; older financial JSON stays untouched but is no longer read. - Do not delete data or reset volumes to fix unavailable reports. Request preparation explicitly and reload to inspect the result. - Background maintenance removes eligible older balance-query jobs, not saved reports. Deletion is not secure erasure; SQLite and backups may retain data. diff --git a/packages/db/src/database.ts b/packages/db/src/database.ts index 0e657f0..97f208b 100644 --- a/packages/db/src/database.ts +++ b/packages/db/src/database.ts @@ -1,4 +1,4 @@ -import { Kysely, sql } from 'kysely'; +import { Kysely, sql, type ControlledTransaction } from 'kysely'; import { constants } from 'node:fs'; import { open } from 'node:fs/promises'; import { dirname } from 'node:path'; @@ -6,6 +6,9 @@ import { DatabaseSync } from 'node:sqlite'; import { nodeSqliteDialect } from './node-sqlite-dialect.js'; import { reportSchemaVersion, type ReportDatabase } from './schema.js'; +export type ReportConnection = Kysely; +export type ReportTransaction = ControlledTransaction; + async function checkDirectory(directory: string): Promise { const folder = await open( directory, diff --git a/packages/db/src/index.ts b/packages/db/src/index.ts index cead1ce..47d6976 100644 --- a/packages/db/src/index.ts +++ b/packages/db/src/index.ts @@ -1,3 +1,4 @@ export { createReportDatabase, openReadOnlyReportDatabase } from './database.js'; +export type { ReportConnection, ReportTransaction } from './database.js'; export { runMigrations } from './migrate.js'; export type { ReportDatabase } from './schema.js'; diff --git a/packages/db/src/reports/queries.test.ts b/packages/db/src/reports/queries.test.ts index 22b58f2..bed68d6 100644 --- a/packages/db/src/reports/queries.test.ts +++ b/packages/db/src/reports/queries.test.ts @@ -1,6 +1,7 @@ import { mkdtemp, mkdir, readFile, rm, stat, symlink, writeFile } from 'node:fs/promises'; import { tmpdir } from 'node:os'; import { join } from 'node:path'; +import { DatabaseSync } from 'node:sqlite'; import { sql, type Kysely } from 'kysely'; import { afterEach, beforeEach, expect, it } from 'vitest'; import { @@ -12,6 +13,7 @@ import { import { readAccountRows, readReportData, writeAccountRows, writeReportData } from './queries.js'; import type { StoredReportData } from './data.js'; import { + amount, ambiguous, preparedOverview, journal, @@ -51,12 +53,64 @@ afterEach(async () => { it('stores exact prepared payloads and native row order using the migrated schema', async () => { const database = await createDatabase(); const rows = preparedAccountRows(52).reverse(); + const rich = rows[0]; + // Preserve producer-owned prices and basis, including unavailable native conversions. + rich.transaction.postings[1].amounts[0] = { + ...amount('75.00'), + commodity: 'TEST', + cost: { kind: 'UnitCost', amount: amount('0.000000000000000000012300') }, + costBasis: { + amount: { ...amount('9007199254740993.123456789'), commodity: 'EUR' }, + date: '2026-04-01', + label: 'retained basis label' + }, + style: { + commoditySide: 'L', + commoditySpaced: false, + decimalMark: '.', + digitGroups: [',', [3]], + precision: 2, + rounding: 'NoRounding' + } + }; + rich.transaction.postings[1].transactedAmounts = [ + { commodity: 'EUR', quantity: '0.000000000000000000922500' } + ]; + rich.transaction.postings[2].tags.push(['_conversion-posting', '']); + rich.transaction.postings[2].transactedAmounts = null; + rich.investment = { + securities: [ + { + commodity: 'TEST', + action: 'unknown', + change: [{ commodity: 'TEST', quantity: '75.00' }], + postingIndexes: [1], + prices: [amount('0.000000000000000000012300')], + transactedAmounts: rich.transaction.postings[1].transactedAmounts + } + ], + cash: { + direction: 'unknown', + change: [], + balance: [], + unclassifiedChange: rich.change, + unclassifiedBalance: rich.balance + }, + cashPostingIndexes: [], + incomePostingIndexes: [], + expensePostingIndexes: [1, 2], + gainPostingIndexes: [], + issues: ['unclassified-commodity', 'unavailable-transacted-values'] + }; + const otherRows = preparedAccountRows(1, 'case-distinct account'); + otherRows[0].transaction.postings[0].account = 'assets:Cash'; const data: StoredReportData = { journal, overview: preparedOverview(), accounts: [ { id: 'empty001', ...ready('assets:empty', 0) }, { id: 'cash0001', ...ready('assets:cash', rows.length) }, + { id: 'cash0002', ...ready('assets:Cash', otherRows.length) }, { id: 'ambig001', ...ambiguous() } ] }; @@ -64,6 +118,7 @@ it('stores exact prepared payloads and native row order using the migrated schem try { await writeAccountRows(transaction, 'assets:cash', 0, rows.slice(0, 51)); await writeAccountRows(transaction, 'assets:cash', 51, rows.slice(51)); + await writeAccountRows(transaction, 'assets:Cash', 0, otherRows); await writeReportData(transaction, data); await transaction.commit().execute(); } finally { @@ -78,13 +133,39 @@ it('stores exact prepared payloads and native row order using the migrated schem expect(await readReportData(reader)).toEqual(data); expect(await readAccountRows(reader, data.accounts[0])).toEqual([]); expect(await readAccountRows(reader, data.accounts[1])).toEqual(rows); - expect(await readAccountRows(reader, data.accounts[2])).toBeNull(); + expect(await readAccountRows(reader, data.accounts[2])).toEqual(otherRows); + expect(await readAccountRows(reader, data.accounts[3])).toBeNull(); await expect( writeAccountRows(reader, 'assets:cash', rows.length, rows.slice(0, 1)) ).rejects.toThrow(); expect((await stat(path)).mode & 0o777).toBe(0o600); }); +it('rejects a persisted row with an out-of-range posting reference', async () => { + const database = await createDatabase(); + const account = { id: 'cash0001', ...ready() }; + const rows = preparedAccountRows(); + await writeReportData(database, { journal, overview: preparedOverview(), accounts: [account] }); + await writeAccountRows(database, account.name, 0, rows); + await closeDatabase(database); + + // Valid SQLite/JSON is not sufficient: saved posting references must still be usable. + const native = new DatabaseSync(path); + try { + const badRow = { ...rows[0], accountPostingIndexes: [rows[0].transaction.postings.length] }; + native + .prepare('UPDATE account_activity SET row_json = ? WHERE account_name = ? AND ordinal = 0') + .run(JSON.stringify(badRow), account.name); + } finally { + native.close(); + } + const reader = await openReadOnlyReportDatabase(path); + if (!reader) throw new Error('Expected the current report schema.'); + databases.add(reader); + expect((await readReportData(reader)).accounts).toEqual([account]); + await expect(readAccountRows(reader, account)).rejects.toThrow('Invalid account data.'); +}); + it('does not create missing storage or replace an existing report file', async () => { const missing = join(directory, 'missing'); await expect(openReadOnlyReportDatabase(join(missing, 'reports.sqlite'))).rejects.toThrow(); diff --git a/packages/snapshots/README.md b/packages/snapshots/README.md index ca30719..9c67374 100644 --- a/packages/snapshots/README.md +++ b/packages/snapshots/README.md @@ -1,7 +1,7 @@ # Saved reports -Stores complete private report sets and recorded journal checks. Writes are staged; -readers keep one selected set for a consistent request. Checks are independent of +Stores complete private SQLite report sets. Writes are staged; readers keep one +selected set for a consistent request. Recorded journal checks are separate from financial publication and do not prove freshness. Reports are derived data, not journal backups. See diff --git a/packages/snapshots/src/reports/publication.test.ts b/packages/snapshots/src/reports/publication.test.ts index 2cb5fe7..ddb3a60 100644 --- a/packages/snapshots/src/reports/publication.test.ts +++ b/packages/snapshots/src/reports/publication.test.ts @@ -1,12 +1,10 @@ -import type { - OverviewData, - PreparedAccountSummary, - PreparedAccountRow -} from '@hledger-reports/hledger'; +import * as databases from '@hledger-reports/db'; +import * as reports from '@hledger-reports/db/reports'; import * as crypto from 'node:crypto'; import * as fs from 'node:fs/promises'; import { tmpdir } from 'node:os'; import { basename, join } from 'node:path'; +import { DatabaseSync } from 'node:sqlite'; import { afterEach, beforeEach, expect, it, vi } from 'vitest'; import { createReportStage, @@ -14,17 +12,13 @@ import { readOverview, readReportPublication } from '../index.js'; -import type { StoredAccount } from '../index.js'; +import type { ReportPublication, StoredAccount } from '../index.js'; import { - amount, ambiguous, preparedOverview as fixture, journal, - range, ready, - preparedAccountRows as rows, - sourcePosition, - undated + preparedAccountRows as rows } from '../../../db/src/reports/data.fixture.js'; vi.mock('node:fs/promises', async (original) => ({ @@ -33,16 +27,20 @@ vi.mock('node:fs/promises', async (original) => ({ vi.mock('node:crypto', async (original) => ({ ...(await original()) })); +vi.mock('@hledger-reports/db', async (original) => ({ + ...(await original()) +})); +vi.mock('@hledger-reports/db/reports', async (original) => ({ + ...(await original()) +})); -const currentReportFormat = 8; let root: string; let snapshots: string; const older = { id: 'older', createdAt: new Date('2026-04-30T23:59:59.999Z') }; const newer = { id: 'newer', createdAt: new Date('2026-05-01T00:00:00.000Z') }; -const filename = (run = older) => - `${String(run.createdAt.getTime()).padStart(16, '0')}-${run.id}.json`; const safeError = (message: string) => (error: Error) => error.message === message && error.cause === undefined; + async function publish( run = older, data = fixture(), @@ -52,21 +50,15 @@ async function publish( await stage.writeRows('assets:cash', rows()); return stage.publish(data, [ready()], journal); } + async function selected() { const publication = await readReportPublication(snapshots); if (!publication) throw new Error('Missing test publication'); return publication; } -async function legacy(contents: unknown, run = older) { - const directory = join(snapshots, 'dashboard'); - await fs.mkdir(directory, { recursive: true }); - const path = join(directory, filename(run)); - await fs.writeFile(path, typeof contents === 'string' ? contents : JSON.stringify(contents)); - return path; -} beforeEach(async () => { - root = await fs.mkdtemp(join(tmpdir(), 'overview-snapshot-')); + root = await fs.mkdtemp(join(tmpdir(), 'report-publication-')); snapshots = join(root, 'snapshots'); }); afterEach(async () => { @@ -74,172 +66,99 @@ afterEach(async () => { await fs.rm(root, { recursive: true, force: true }); }); -it('returns null for absent/empty stores without creating files', async () => { +it('returns null for absent and empty stores without creating reports', async () => { expect(await readOverview(snapshots)).toBeNull(); expect(await fs.readdir(root)).toEqual([]); await fs.mkdir(join(snapshots, '.sets'), { recursive: true }); expect(await readReportPublication(snapshots)).toBeNull(); + expect(await fs.readdir(join(snapshots, '.sets'))).toEqual([]); }); -it('publishes one readable overview and one whole-account JSON, with private modes', async () => { - expect(await publish()).toBeUndefined(); - expect(await readOverview(snapshots)).toEqual(fixture()); - const publication = await selected(); - const account = publication.accounts[0]; - expect(account).toMatchObject(ready()); - expect(account.id).toMatch(/^[a-z0-9]{8}$/); - expect( - JSON.parse(await fs.readFile(join(publication.directory, 'overview.json'), 'utf8')) - ).toEqual({ - format: currentReportFormat, - overview: fixture(), - journal, - accounts: [account] +it('commits and closes a complete private database before making its directory visible', async () => { + await publish(); + const pinned = await selected(); + const createDatabase = databases.createReportDatabase; + const closed = vi.fn(); + vi.spyOn(databases, 'createReportDatabase').mockImplementation(async (path) => { + const database = await createDatabase(path); + const destroy = database.destroy.bind(database); + vi.spyOn(database, 'destroy').mockImplementation(async () => { + await destroy(); + closed(); + }); + return database; }); - expect(await fs.readlink(join(snapshots, 'current'))).toBe( - `.sets/${basename(publication.directory)}` - ); - expect(await fs.readdir(snapshots)).toEqual(['.sets', 'current']); - expect((await fs.readdir(publication.directory)).sort()).toEqual([ - 'overview.json', - 'transactions' - ]); - expect(await fs.readdir(join(publication.directory, 'transactions'))).toEqual([ - `${account.id}.json` - ]); + const stage = await createReportStage(snapshots, newer, pinned.accounts); + const transactions = rows(2); + await stage.writeRows('assets:cash', transactions.slice(0, 1)); + await stage.writeRows('assets:cash', transactions.slice(1)); + const rename = fs.rename; + const inspected = vi.fn(); + vi.spyOn(fs, 'rename').mockImplementation(async (source, destination) => { + if (String(source).endsWith('.tmp') && String(destination).includes('/.sets/')) { + expect(await selected()).toEqual(pinned); + expect(closed).toHaveBeenCalledOnce(); + const database = new DatabaseSync(join(String(source), 'reports.sqlite'), { + readOnly: true + }); + try { + const record = database.prepare('SELECT journal, overview_json FROM publication').get()!; + expect(record.journal).toBe(journal); + expect(JSON.parse(record.overview_json as string)).toEqual(fixture()); + expect( + database + .prepare('SELECT row_json FROM account_activity ORDER BY ordinal') + .all() + .map((row) => JSON.parse(row.row_json as string)) + ).toEqual(transactions); + } finally { + database.close(); + } + inspected(); + } + await rename(source, destination); + }); + await stage.publish(fixture(), [ready('assets:cash', 2)], journal); + expect(inspected).toHaveBeenCalledOnce(); + const publication = await selected(); + expect(publication.overview).toEqual(fixture()); + expect(publication.journal).toBe(journal); + const account = publication.accounts[0]; + expect(account).toMatchObject(ready('assets:cash', 2)); expect(await readAccountTransactions(publication, account.id)).toEqual({ id: account.id, name: account.name, - transactions: rows() - }); - for (const directory of [ - snapshots, - join(snapshots, '.sets'), - publication.directory, - join(publication.directory, 'transactions') - ]) - expect((await fs.stat(directory)).mode & 0o777).toBe(0o700); - for (const path of [ - join(publication.directory, 'overview.json'), - join(publication.directory, 'transactions', `${account.id}.json`) - ]) - expect((await fs.stat(path)).mode & 0o777).toBe(0o600); -}); - -it('round-trips the complete rich record and private provenance in the known account file', async () => { - const stage = await createReportStage(snapshots, older, [{ ...ready(), id: 'cash0001' }]); - const transactions = rows(); - // Producer-owned original prices, basis and valuation survive without reinterpretation. - transactions[0].transaction.postings[1].amounts[0] = { - ...amount('75.00'), - commodity: 'TEST', - cost: { kind: 'UnitCost', amount: amount('0.000000000000000000012300') }, - costBasis: { - amount: { ...amount('9007199254740993.123456789'), commodity: 'EUR' }, - date: '2026-04-01', - label: 'retained basis label' - }, - style: { - commoditySide: 'L', - commoditySpaced: false, - decimalMark: '.', - digitGroups: [',', [3]], - precision: 2, - rounding: 'NoRounding' - } - }; - transactions[0].transaction.postings[1].transactedAmounts = [ - { commodity: 'EUR', quantity: '0.000000000000000000922500' }, - { commodity: '$', quantity: '9007199254740993.123456789' } - ]; - // A native-omitted conversion posting is unavailable, not a fabricated zero bag. - transactions[0].transaction.postings[2].tags.push(['_conversion-posting', '']); - transactions[0].transaction.postings[2].transactedAmounts = null; - transactions[0].investment = { - securities: [ - { - commodity: 'TEST', - action: 'unknown', - change: [{ commodity: 'TEST', quantity: '75.00' }], - postingIndexes: [1], - prices: [amount('0.000000000000000000012300')], - transactedAmounts: transactions[0].transaction.postings[1].transactedAmounts - } - ], - cash: { - direction: 'unknown', - change: [], - balance: [], - unclassifiedChange: transactions[0].change, - unclassifiedBalance: transactions[0].balance - }, - cashPostingIndexes: [], - incomePostingIndexes: [], - expensePostingIndexes: [1, 2], - gainPostingIndexes: [], - issues: ['unclassified-commodity', 'unavailable-transacted-values'] - }; - await stage.writeRows('assets:cash', transactions); - await stage.publish(fixture(), [ready()], journal); - const publication = await selected(); - const path = join(publication.directory, 'transactions', 'cash0001.json'); - const raw = await fs.readFile(path, 'utf8'); - expect(JSON.parse(raw)).toEqual({ id: 'cash0001', name: 'assets:cash', transactions }); - expect((await readAccountTransactions(publication, 'cash0001')).transactions).toEqual( transactions + }); + expect(await fs.readdir(publication.directory)).toEqual(['reports.sqlite']); + expect(await fs.readlink(join(snapshots, 'current'))).toBe( + `.sets/${basename(publication.directory)}` ); - expect(raw).toContain(sourcePosition.file); - expect( - JSON.parse(raw).transactions[0].transaction.postings.map( - (posting: { amounts: { quantity: string }[] }) => posting.amounts[0].quantity - ) - ).toEqual(['-100.00', '75.00', '25.00']); + for (const directory of [snapshots, join(snapshots, '.sets'), publication.directory]) + expect((await fs.stat(directory)).mode & 0o777).toBe(0o700); + expect((await fs.stat(join(publication.directory, 'reports.sqlite'))).mode & 0o777).toBe(0o600); }); -it('retains all rows in one file; callers slice 50 after reading it', async () => { - const stage = await createReportStage(snapshots, older); - const all = rows(137); - for (let i = 0; i < all.length; i += 50) - await stage.writeRows('assets:cash', all.slice(i, i + 50)); - await stage.publish(fixture(), [ready('assets:cash', all.length)], journal); - const publication = await selected(); - const result = await readAccountTransactions(publication, publication.accounts[0].id); - expect(result.transactions).toEqual(all); - expect(await fs.readdir(join(publication.directory, 'transactions'))).toHaveLength(1); -}); - -it('represents empty-ready and ambiguous accounts distinctly, including a missing zero-row callback', async () => { +it('distinguishes empty-ready accounts from ambiguous access without requiring empty callbacks', async () => { const stage = await createReportStage(snapshots, older); await stage.writeRows('empty callback', []); - const accounts: PreparedAccountSummary[] = [ - undated('empty'), - undated('empty callback'), - ambiguous() - ]; - const overview: OverviewData = { - ...fixture(), - asOf: null, - incomeExpenses: { months: [] }, - journalCoverage: { transactionCount: 0, firstDate: null, lastDate: null }, - ranges: null, - balanceSheet: null, - netWorth: { currency: null, points: [] } - }; - await stage.publish(overview, accounts, journal); - expect(await readOverview(snapshots)).toEqual(overview); + await stage.publish( + fixture(), + [ready('empty', 0), ready('empty callback', 0), ambiguous()], + journal + ); const publication = await selected(); - for (const account of publication.accounts) { + for (const account of publication.accounts) expect(await readAccountTransactions(publication, account.id)).toEqual({ id: account.id, name: account.name, transactions: account.access === 'ready' ? [] : null }); - } }); -it('preserves Unicode, delimiters and exact identity as data; IDs are random and collision-checked against all prior/new IDs', async () => { +it('reuses IDs by exact account identity and avoids collisions with removed and newly assigned IDs', async () => { const generated = [0, 1]; - const random = vi.spyOn(crypto, 'randomInt').mockImplementation(() => generated.shift()!); + vi.spyOn(crypto, 'randomInt').mockImplementation(() => generated.shift()!); const first = await createReportStage(snapshots, older); await first.publish(fixture(), [ready('removed', 0), ready('資産/../銀行:Ω', 0)], journal); const previous = await selected(); @@ -261,77 +180,37 @@ it('preserves Unicode, delimiters and exact identity as data; IDs are random and ]); for (const account of publication.accounts) expect((await readAccountTransactions(publication, account.id)).name).toBe(account.name); - expect((await fs.readdir(join(publication.directory, 'transactions'))).sort()).toEqual([ - '00000001.json', - '00000002.json', - '00000003.json' - ]); - expect(random).toHaveBeenCalledWith(36 ** 8); }); -it('keeps the old set pinned during and after another publication and same-run recovery', async () => { +it('keeps completed generations pinned through same-request recovery and discard', async () => { await publish(); const pinned = await selected(); const stage = await createReportStage(snapshots, older, pinned.accounts); await stage.writeRows('assets:cash', rows(1, 'recovered')); - expect(await readReportPublication(snapshots)).toEqual(pinned); + expect(await selected()).toEqual(pinned); await stage.publish({ ...fixture(), preparedAt: 'recovered' }, [ready()], journal); await stage.discard(); - const sets = (await fs.readdir(join(snapshots, '.sets'))).sort(); + const sets = await fs.readdir(join(snapshots, '.sets')); expect(sets).toHaveLength(2); - expect(new Set(sets).size).toBe(2); expect((await readAccountTransactions(pinned, pinned.accounts[0].id)).transactions).toEqual( rows() ); - // Every completed execution remains readable, regardless of the same-request tie-break. - const other = sets.find((set) => set !== basename(pinned.directory))!; - expect( - JSON.parse( - await fs.readFile( - join(snapshots, '.sets', other, 'transactions', `${pinned.accounts[0].id}.json`), - 'utf8' - ) - ).transactions - ).toEqual(rows(1, 'recovered')); + // Same-request selection may choose either UUID; both completed executions remain readable. + const recovered = { + ...pinned, + directory: join(snapshots, '.sets', sets.find((set) => set !== basename(pinned.directory))!) + }; + expect((await readAccountTransactions(recovered, pinned.accounts[0].id)).transactions).toEqual( + rows(1, 'recovered') + ); }); -it('commits the entire set only at directory rename and ignores hidden stages', async () => { - await publish(); - const pinned = await selected(); - const stage = await createReportStage(snapshots, newer, pinned.accounts); - await stage.writeRows('assets:cash', rows(1, 'next')); - const rename = fs.rename; - let inspected = false; - vi.spyOn(fs, 'rename').mockImplementation(async (source, destination) => { - if (String(source).endsWith('.tmp') && String(destination).includes('/.sets/')) { - expect(await readReportPublication(snapshots)).toEqual(pinned); - const record = JSON.parse(await fs.readFile(join(String(source), 'overview.json'), 'utf8')); - expect(record.overview.preparedAt).toBe('next'); - expect(record.journal).toBe(journal); - expect( - JSON.parse( - await fs.readFile( - join(String(source), 'transactions', `${record.accounts[0].id}.json`), - 'utf8' - ) - ).transactions - ).toEqual(rows(1, 'next')); - inspected = true; - } - await rename(source, destination); - }); - await stage.publish({ ...fixture(), preparedAt: 'next' }, [ready()], journal); - expect(inspected).toBe(true); - expect((await readOverview(snapshots))!.preparedAt).toBe('next'); -}); - -it('selects request creation order despite reversed dates, old recovery and an untrusted current alias', async () => { +it('selects request creation order, not report dates or the current alias', async () => { const next = { ...fixture(), asOf: '2025-01-01', preparedAt: '2025-01-02' }; await publish(newer, next); const newest = await selected(); - expect(await publish(older)).toBeUndefined(); + await publish(older); expect(await readOverview(snapshots)).toEqual(next); - expect(await fs.readlink(join(snapshots, 'current'))).toBe(`.sets/${basename(newest.directory)}`); await fs.unlink(join(snapshots, 'current')); await fs.symlink('/not-a-trusted-publication', join(snapshots, 'current')); expect(await readOverview(snapshots)).toEqual(next); @@ -339,21 +218,11 @@ it('selects request creation order despite reversed dates, old recovery and an u expect(await fs.readlink(join(snapshots, 'current'))).toBe(`.sets/${basename(newest.directory)}`); }); -it('uses run ID as the request tie-break for same timestamps', async () => { - await publish({ ...newer, id: 'z' }); - expect( - await publish({ ...newer, id: 'a' }, { ...fixture(), preparedAt: 'not selected' }) - ).toBeUndefined(); - expect(await readOverview(snapshots)).toEqual(fixture()); -}); - -it('retains baseline request ordering for IDs that prefix other IDs', async () => { - const prefix = { ...newer, id: 'a' }; - const extension = { ...newer, id: 'a-z' }; +it('preserves request ordering when one run ID prefixes another', async () => { const current = { ...fixture(), preparedAt: 'selected prefix' }; - // The request separator sorts after a hyphen: the prefix request wins. - await publish(prefix, current); - expect(await publish(extension)).toBeUndefined(); + await publish({ ...newer, id: 'a' }, current); + // The request separator sorts after a hyphen, so the prefix request wins. + await publish({ ...newer, id: 'a-z' }); expect(await readOverview(snapshots)).toEqual(current); const publication = await selected(); expect(await fs.readlink(join(snapshots, 'current'))).toBe( @@ -361,7 +230,7 @@ it('retains baseline request ordering for IDs that prefix other IDs', async () = ); }); -it('repairs a delayed older alias update after a newer set appears', async () => { +it('repairs a delayed older alias replacement after a newer publication appears', async () => { const rename = fs.rename; let release!: () => void; let reached!: () => void; @@ -384,105 +253,186 @@ it('repairs a delayed older alias update after a newer set appears', async () => } finally { release(); } - expect(await old).toBeUndefined(); + await old; expect(await readOverview(snapshots)).toEqual(next); const newest = await selected(); expect(await fs.readlink(join(snapshots, 'current'))).toBe(`.sets/${basename(newest.directory)}`); }); -it('retains all completed sets from concurrent distinct requests', async () => { - await Promise.all( - Array.from({ length: 5 }, (_, i) => - publish({ id: `run_${i}`, createdAt: new Date(newer.createdAt.getTime() + i) }) - ) +it('closes a failed database setup and removes only its private stage', async () => { + await publish(); + const pinned = await selected(); + const createDatabase = databases.createReportDatabase; + const closed = vi.fn(); + vi.spyOn(databases, 'createReportDatabase').mockImplementation(async (path) => { + const database = await createDatabase(path); + const destroy = database.destroy.bind(database); + vi.spyOn(database, 'destroy').mockImplementation(async () => { + await destroy(); + closed(); + }); + return database; + }); + vi.spyOn(databases, 'runMigrations').mockRejectedValueOnce(new Error('private setup failure')); + await expect(createReportStage(snapshots, newer)).rejects.toSatisfy( + safeError('Unable to prepare reports') ); - expect(await fs.readdir(join(snapshots, '.sets'))).toHaveLength(5); - const newest = await selected(); - expect(basename(newest.directory)).toContain('-run_4-'); - expect(await fs.readlink(join(snapshots, 'current'))).toBe(`.sets/${basename(newest.directory)}`); + expect(closed).toHaveBeenCalledOnce(); + expect(await selected()).toEqual(pinned); + expect(await fs.readdir(join(snapshots, '.sets'))).toEqual([basename(pinned.directory)]); }); -it.each( - ( - [ - [ready('missing', 1)], - [ready('assets:cash', 2)], - [], - [ambiguous('assets:cash')], - [ready(), ready()], - [ready('assets:cash', -1)] - ] as PreparedAccountSummary[][] - ).map((accounts) => ({ accounts })) -)('rejects summary mismatches and cleans only its hidden stage (%j)', async ({ accounts }) => { +it('preserves private staging when failed setup cannot confirm a database close', async () => { + await publish(); + const pinned = await selected(); + const destroy = vi.fn(async () => { + throw new Error('private close failure'); + }); + vi.spyOn(databases, 'createReportDatabase').mockResolvedValueOnce({ + destroy + } as unknown as databases.ReportConnection); + vi.spyOn(databases, 'runMigrations').mockRejectedValueOnce(new Error('private setup failure')); + const remove = vi.spyOn(fs, 'rm'); + await expect(createReportStage(snapshots, newer)).rejects.toSatisfy( + safeError('Unable to prepare reports') + ); + expect(destroy).toHaveBeenCalledOnce(); + expect(remove).not.toHaveBeenCalled(); + expect(await selected()).toEqual(pinned); + expect( + (await fs.readdir(join(snapshots, '.sets'))).filter((name) => name.startsWith('.')) + ).toHaveLength(1); +}); + +it('retains unclosed staging and resumes cleanup after a failed row write', async () => { + await publish(); + const pinned = await selected(); + const sets = join(snapshots, '.sets'); + await fs.mkdir(join(sets, '.foreign.tmp')); + const events: string[] = []; + let release!: () => void; + let entered!: () => void; + const held = new Promise((resolve) => (release = resolve)); + const closing = new Promise((resolve) => (entered = resolve)); + const transaction = { + isCommitted: false, + isRolledBack: false, + rollback: () => ({ + execute: async () => { + events.push('rollback'); + transaction.isRolledBack = true; + } + }) + }; + const destroy = vi.fn(async () => { + events.push('closing'); + entered(); + await held; + events.push('closed'); + }); + destroy.mockImplementationOnce(async () => { + events.push('close rejected'); + throw new Error('private close failure'); + }); + vi.spyOn(databases, 'createReportDatabase').mockResolvedValueOnce({ + startTransaction: () => ({ execute: async () => transaction }), + destroy + } as unknown as databases.ReportConnection); + vi.spyOn(databases, 'runMigrations').mockResolvedValueOnce(undefined); + const write = vi.spyOn(reports, 'writeAccountRows').mockResolvedValueOnce(undefined); + write.mockRejectedValueOnce(new Error('private write failure')); + const remove = vi.spyOn(fs, 'rm'); + const stage = await createReportStage(snapshots, newer, pinned.accounts); + await stage.writeRows('assets:cash', rows()); + await expect(stage.writeRows('assets:cash', rows())).rejects.toSatisfy( + safeError('Unable to prepare reports') + ); + expect(events).toEqual(['rollback', 'close rejected']); + expect(remove).not.toHaveBeenCalled(); + const retry = stage.discard(); + try { + await closing; + expect(remove).not.toHaveBeenCalled(); + expect((await fs.readdir(sets)).filter((name) => name.startsWith('.'))).toHaveLength(2); + } finally { + release(); + await retry; + } + expect(events).toEqual(['rollback', 'close rejected', 'closing', 'closed']); + await expect(stage.writeRows('assets:cash', rows())).rejects.toThrow(); + expect(write).toHaveBeenCalledTimes(2); + expect(await selected()).toEqual(pinned); + expect((await fs.readdir(sets)).sort()).toEqual( + ['.foreign.tmp', basename(pinned.directory)].sort() + ); +}); + +it.each([ + { condition: 'incorrect row counts', summary: ready('assets:cash', 2) }, + { condition: 'written rows declared ambiguous', summary: ambiguous('assets:cash') } +])('rejects $condition without replacing a completed generation', async ({ summary }) => { + await publish(); + const pinned = await selected(); + const stage = await createReportStage(snapshots, newer, pinned.accounts); + await stage.writeRows('assets:cash', rows()); + await expect(stage.publish(fixture(), [summary], journal)).rejects.toSatisfy( + safeError('Unable to prepare reports') + ); + expect(await selected()).toEqual(pinned); + expect(await fs.readdir(join(snapshots, '.sets'))).toEqual([basename(pinned.directory)]); +}); + +it('cleans only its stage when the publication directory rename fails', async () => { await publish(); const pinned = await selected(); const sets = join(snapshots, '.sets'); await fs.mkdir(join(sets, '.foreign.tmp')); const stage = await createReportStage(snapshots, newer, pinned.accounts); await stage.writeRows('assets:cash', rows()); - await expect(stage.publish(fixture(), accounts, journal)).rejects.toSatisfy( + vi.spyOn(fs, 'rename').mockRejectedValue(new Error(`${snapshots}: private rename failure`)); + await expect(stage.publish(fixture(), [ready()], journal)).rejects.toSatisfy( safeError('Unable to prepare reports') ); - expect(await readReportPublication(snapshots)).toEqual(pinned); + await stage.discard(); + expect(await selected()).toEqual(pinned); expect((await fs.readdir(sets)).sort()).toEqual( ['.foreign.tmp', basename(pinned.directory)].sort() ); }); -it('rejects a return to an already-closed account and closes before discarding', async () => { +it('rejects noncontiguous writes and retries removal without closing the database again', async () => { + const createDatabase = databases.createReportDatabase; + const closed = vi.fn(); + vi.spyOn(databases, 'createReportDatabase').mockImplementation(async (path) => { + const database = await createDatabase(path); + const destroy = database.destroy.bind(database); + vi.spyOn(database, 'destroy').mockImplementation(async () => { + await destroy(); + closed(); + }); + return database; + }); const stage = await createReportStage(snapshots, older); await stage.writeRows('a', rows()); await stage.writeRows('b', rows()); - await expect(stage.writeRows('a', rows())).rejects.toThrow('Unable to prepare reports'); + const remove = fs.rm; + vi.spyOn(fs, 'rm') + .mockRejectedValueOnce(new Error('private removal failure')) + .mockImplementation(remove); + await expect(stage.writeRows('a', rows())).rejects.toSatisfy( + safeError('Unable to prepare reports') + ); + expect(closed).toHaveBeenCalledOnce(); + expect(await fs.readdir(join(snapshots, '.sets'))).toHaveLength(1); await stage.discard(); + await stage.discard(); + expect(closed).toHaveBeenCalledOnce(); + await expect(stage.writeRows('b', rows())).rejects.toThrow(); + expect(await readOverview(snapshots)).toBeNull(); expect(await fs.readdir(join(snapshots, '.sets'))).toEqual([]); }); -it('discards idempotently without publishing, and cannot reuse a discarded stage', async () => { - const stage = await createReportStage(snapshots, older); - await stage.writeRows('assets:cash', rows()); - await stage.discard(); - await stage.discard(); - expect(await readOverview(snapshots)).toBeNull(); - await expect(stage.writeRows('assets:cash', rows())).rejects.toThrow('Unable to prepare reports'); -}); - -it('retains a completed set after post-commit alias selection failure', async () => { - const stage = await createReportStage(snapshots, older); - await stage.writeRows('assets:cash', rows()); - const rename = fs.rename; - vi.spyOn(fs, 'rename').mockImplementation(async (source, destination) => { - await rename(source, destination); - if (String(destination).includes('/.sets/')) - vi.spyOn(fs, 'readdir').mockRejectedValue(new Error('private path')); - }); - await expect(stage.publish(fixture(), [ready()], journal)).rejects.toSatisfy( - safeError('Unable to prepare reports') - ); - await stage.discard(); - vi.restoreAllMocks(); - expect(await fs.readdir(join(snapshots, '.sets'))).toHaveLength(1); - const publication = await selected(); - expect( - (await readAccountTransactions(publication, publication.accounts[0].id)).transactions - ).toEqual(rows()); -}); - -it('cleans only its stage on rename failure, preserving legacy data and previous completion', async () => { - const oldPath = await legacy(fixture()); - await publish(); - const pinned = await selected(); - const stage = await createReportStage(snapshots, newer); - await stage.writeRows('assets:cash', rows()); - vi.spyOn(fs, 'rename').mockRejectedValue(new Error(`${snapshots}: private contents`)); - await expect(stage.publish(fixture(), [ready()], journal)).rejects.toSatisfy( - safeError('Unable to prepare reports') - ); - expect(await fs.readdir(join(snapshots, '.sets'))).toEqual([basename(pinned.directory)]); - expect(await fs.readFile(oldPath, 'utf8')).toBe(JSON.stringify(fixture())); -}); - -it('does not delete a stage it could not create exclusively', async () => { +it('does not remove a stage it could not create exclusively', async () => { const uuid = '00000000-0000-4000-8000-000000000000'; vi.spyOn(crypto, 'randomUUID').mockReturnValue(uuid); const foreign = join(snapshots, '.sets', `.${uuid}.tmp`); @@ -492,616 +442,187 @@ it('does not delete a stage it could not create exclusively', async () => { expect(await fs.readFile(join(foreign, 'foreign'), 'utf8')).toBe('untouched'); }); -it.each(['../escape', '', 'x/y', 'x\\y', 'x\n', 'x'.repeat(129)])( - 'rejects unsafe run IDs before creating files (%j)', - async (id) => { - await expect(createReportStage(snapshots, { ...older, id })).rejects.toThrow( - 'Unable to prepare reports' - ); - expect(await fs.readdir(root)).toEqual([]); - } -); -it.each([new Date('invalid'), new Date(-1)])( - 'rejects invalid timestamps (%s)', - async (createdAt) => { - await expect(createReportStage(snapshots, { ...older, createdAt })).rejects.toThrow( - 'Unable to prepare reports' - ); - expect(await fs.readdir(root)).toEqual([]); - } -); - -it('rejects BigInt/cyclic data via ordinary serialization without harming the previous publication', async () => { +it('publishes without financial readback and keeps the new generation after alias failure', async () => { await publish(); const pinned = await selected(); - for (const extra of [ - 1n, - (() => { - const value: unknown[] = []; - value.push(value); - return value; - })() - ]) { - const stage = await createReportStage(snapshots, newer); - await expect( - stage.writeRows('a', [{ ...rows()[0], extra }] as unknown as PreparedAccountRow[]) - ).rejects.toSatisfy(safeError('Unable to prepare reports')); - } - expect(await readReportPublication(snapshots)).toEqual(pinned); -}); - -it('publishes headers, appended rows and JSON without a byte-length policy', async () => { - // A controlled size double exercises the former checks without a large payload. - const size = vi.spyOn(Buffer, 'byteLength').mockReturnValue(64 * 1024 * 1024 + 1); - const stage = await createReportStage(snapshots, older); - const transactions = rows(2); - await stage.writeRows('assets:cash', transactions.slice(0, 1)); - await stage.writeRows('assets:cash', transactions.slice(1)); - await stage.publish(fixture(), [ready('assets:cash', 2), ready('empty', 0)], journal); - await stage.discard(); - size.mockRestore(); - const publication = await selected(); - expect(publication.overview).toEqual(fixture()); - for (const account of publication.accounts) - expect((await readAccountTransactions(publication, account.id)).transactions).toEqual( - account.name === 'assets:cash' ? transactions : [] - ); - expect(await fs.readdir(join(snapshots, '.sets'))).toEqual([basename(publication.directory)]); -}); - -it('reads a pinned account above the former byte bound using a size double', async () => { - await publish(); - const publication = await selected(); - const account = publication.accounts[0]; - const path = join(publication.directory, 'transactions', `${account.id}.json`); - const raw = await fs.readFile(path, 'utf8'); - const reportedSize = 64 * 1024 * 1024 + 1; - // Model a large, consistent read without allocating or storing a large fixture. - const toString = vi.fn().mockReturnValue(raw); - const buffer = { length: reportedSize + 1, toString } as unknown as Buffer; - const alloc = Buffer.alloc; - const allocation = vi - .spyOn(Buffer, 'alloc') - .mockImplementation((size, ...args) => - size === reportedSize + 1 ? buffer : alloc(size, ...args) - ); - const read = vi - .fn() - .mockResolvedValueOnce({ bytesRead: reportedSize, buffer }) - .mockResolvedValueOnce({ bytesRead: 0, buffer }); - const close = vi.fn(); - const open = fs.open; - vi.spyOn(fs, 'open').mockImplementation(async (openedPath, ...args) => { - const file = await open(openedPath, ...args); - if (openedPath === path) { - const stat = await file.stat(); - vi.spyOn(file, 'stat').mockResolvedValue(Object.assign(stat, { size: reportedSize })); - vi.spyOn(file, 'read').mockImplementation(read); - const originalClose = file.close.bind(file); - vi.spyOn(file, 'close').mockImplementation(async () => { - close(); - await originalClose(); - }); - } - return file; - }); - expect((await readAccountTransactions(publication, account.id)).transactions).toEqual(rows()); - expect(allocation).toHaveBeenCalledWith(reportedSize + 1); - expect(read).toHaveBeenNthCalledWith(1, buffer, 0, reportedSize + 1, null); - expect(read).toHaveBeenNthCalledWith(2, buffer, reportedSize, 1, null); - expect(toString).toHaveBeenCalledWith('utf8', 0, reportedSize); - expect(close).toHaveBeenCalledOnce(); -}); - -it.each(['growth', 'shrinkage'])('rejects account file %s after stat', async (change) => { - await publish(); - const publication = await selected(); - const account = publication.accounts[0]; - const path = join(publication.directory, 'transactions', `${account.id}.json`); - const open = fs.open; - const changed = vi.fn(); - vi.spyOn(fs, 'open').mockImplementation(async (openedPath, ...args) => { - const file = await open(openedPath, ...args); - if (openedPath === path) { - const stat = file.stat.bind(file); - vi.spyOn(file, 'stat').mockImplementation(async () => { - const inspected = await stat(); - if (change === 'growth') await fs.appendFile(path, ' '); - else await fs.truncate(path, inspected.size - 1); - changed(); - return inspected; - }); - } - return file; - }); - await expect(readAccountTransactions(publication, account.id)).rejects.toSatisfy( - safeError('Unable to read account transactions') - ); - expect(changed).toHaveBeenCalledOnce(); -}); - -it.each([ - ['header', 1], - ['first rows', 2], - ['appended rows', 3], - ['closing delimiter', 4], - ['empty account JSON', 5], - ['overview JSON', 6] -] as const)('cleans only its stage on actual %s write failure', async (_part, failAt) => { - await publish(); - const pinned = await selected(); - const sets = join(snapshots, '.sets'); - await fs.mkdir(join(sets, '.foreign.tmp')); + const reader = vi + .spyOn(databases, 'openReadOnlyReportDatabase') + .mockRejectedValue(new Error('Publication must not read selected financial data')); const stage = await createReportStage(snapshots, newer, pinned.accounts); - const open = fs.open; - let writes = 0; - const failed = vi.fn(); - vi.spyOn(fs, 'open').mockImplementation(async (path, ...args) => { - const file = await open(path, ...args); - if (String(path).endsWith('.json')) { - const writeFile = file.writeFile.bind(file); - vi.spyOn(file, 'writeFile').mockImplementation(async (...writeArgs) => { - if (++writes === failAt) { - failed(); - throw new Error(`${path}: private write failure`); - } - await writeFile(...writeArgs); - }); - } - return file; - }); - await expect( - (async () => { - await stage.writeRows('assets:cash', rows()); - await stage.writeRows('assets:cash', rows()); - await stage.publish(fixture(), [ready('assets:cash', 2), ready('empty', 0)], journal); - })() - ).rejects.toSatisfy(safeError('Unable to prepare reports')); - expect(failed).toHaveBeenCalledOnce(); - await stage.discard(); - expect(await readReportPublication(snapshots)).toEqual(pinned); - expect((await fs.readdir(sets)).sort()).toEqual( - ['.foreign.tmp', basename(pinned.directory)].sort() - ); -}); - -it.each(['../escape', 'assets:cash', '00000000\n', '0000000', 'A0000000'])( - 'rejects account IDs not in the safe catalog (%j)', - async (id) => { - await publish(); - await expect(readAccountTransactions(await selected(), id)).rejects.toSatisfy( - safeError('Unable to read account transactions') - ); - } -); - -it.each(['id', 'name', 'count', 'null', 'malformed'])( - 'fails closed on mismatched account metadata (%s)', - async (bad) => { - await publish(); - const publication = await selected(); - const account = publication.accounts[0]; - const record = { - id: account.id, - name: account.name, - transactions: rows() as PreparedAccountRow[] | null - }; - if (bad === 'id') record.id = 'zzzzzzzz'; - if (bad === 'name') record.name = 'other'; - if (bad === 'count') record.transactions = []; - if (bad === 'null') record.transactions = null; - await fs.writeFile( - join(publication.directory, 'transactions', `${account.id}.json`), - bad === 'malformed' ? 'private invalid JSON' : JSON.stringify(record) - ); - await expect(readAccountTransactions(publication, account.id)).rejects.toSatisfy( - safeError('Unable to read account transactions') - ); - } -); - -it.each([ - null, - {}, - { ...rows()[0], accountDate: undefined }, - { ...rows()[0], accountDate: null }, - { ...rows()[0], accountDate: 20260430 }, - { ...rows()[0], accountDate: undefined, date: '2026-04-30' }, - { ...rows()[0], transaction: undefined }, - { ...rows()[0], transaction: null }, - { ...rows()[0], transaction: { postings: null } }, - { ...rows()[0], accountPostingIndexes: undefined }, - { ...rows()[0], accountPostingIndexes: null }, - { ...rows()[0], accountPostingIndexes: ['0'] }, - { ...rows()[0], accountPostingIndexes: [-1] }, - { ...rows()[0], accountPostingIndexes: [0.5] }, - { ...rows()[0], accountPostingIndexes: [3] } -])('rejects missing rich-record containers or invalid posting references (%j)', async (row) => { - await publish(); - const publication = await selected(); - const account = publication.accounts[0]; - const path = join(publication.directory, 'transactions', `${account.id}.json`); - const raw = JSON.stringify({ id: account.id, name: account.name, transactions: [row] }); - await fs.writeFile(path, raw); - await expect(readAccountTransactions(publication, account.id)).rejects.toSatisfy( - safeError('Unable to read account transactions') - ); - expect(await fs.readFile(path, 'utf8')).toBe(raw); -}); - -it.each([ - ...[ - ['investment'], - ['investment', 'securities'], - ['investment', 'cash'], - ['investment', 'cash', 'direction'], - ['investment', 'cash', 'change'], - ['investment', 'cash', 'balance'], - ['investment', 'cash', 'unclassifiedChange'], - ['investment', 'cash', 'unclassifiedBalance'], - ['investment', 'cashPostingIndexes'], - ['investment', 'incomePostingIndexes'], - ['investment', 'expensePostingIndexes'], - ['investment', 'gainPostingIndexes'], - ['investment', 'issues'], - ['transaction', 'postings', '0', 'transactedAmounts'] - ].map((path) => ({ path, value: undefined })), - { path: ['investment', 'cash', 'direction'], value: 'invalid' }, - { path: ['investment', 'cash', 'change'], value: {} } -])( - 'rejects missing or malformed current investment/valuation containers ($path: $value)', - async ({ path: fields, value }) => { - await publish(); - const publication = await selected(); - const account = publication.accounts[0]; - const row = rows()[0]; - let container = row as unknown as Record; - for (const field of fields.slice(0, -1)) - container = container[field] as Record; - container[fields.at(-1)!] = value; - const path = join(publication.directory, 'transactions', `${account.id}.json`); - const raw = JSON.stringify({ id: account.id, name: account.name, transactions: [row] }); - await fs.writeFile(path, raw); - await expect(readAccountTransactions(publication, account.id)).rejects.toSatisfy( - safeError('Unable to read account transactions') - ); - expect(await fs.readFile(path, 'utf8')).toBe(raw); - } -); - -it('rejects symlinks at the final set, transactions directory and account file', async () => { - await publish(); - const publication = await selected(); - const account = publication.accounts[0]; - const file = join(publication.directory, 'transactions', `${account.id}.json`); - const saved = join(root, 'saved'); - await fs.rename(file, saved); - await fs.symlink(saved, file); - await expect(readAccountTransactions(publication, account.id)).rejects.toThrow( - 'Unable to read account transactions' - ); - await fs.unlink(file); - await fs.rename(saved, file); - const transactions = join(publication.directory, 'transactions'); - await fs.rename(transactions, saved); - await fs.symlink(saved, transactions); - await expect(readAccountTransactions(publication, account.id)).rejects.toThrow( - 'Unable to read account transactions' - ); - await fs.unlink(transactions); - await fs.rename(saved, transactions); - await fs.rename(publication.directory, saved); - await fs.symlink(saved, publication.directory); - await expect(readOverview(snapshots)).rejects.toThrow('Unable to read overview'); -}); - -it('ignores old dashboard files without scanning or reading them, preserving their bytes', async () => { - const raw = 'not JSON: private old dashboard'; - const path = await legacy(raw, newer); - const open = fs.open; - vi.spyOn(fs, 'open').mockImplementation(async (path, ...args) => { - expect(String(path)).not.toContain('/dashboard'); - return open(path, ...args); - }); - const readdir = fs.readdir; - vi.spyOn(fs, 'readdir').mockImplementation((...args) => { - expect(String(args[0])).not.toContain('/dashboard'); - return readdir(...args); - }); - expect(await readReportPublication(snapshots)).toBeNull(); - expect(await publish()).toBeUndefined(); - expect(await readOverview(snapshots)).toEqual(fixture()); - expect(await fs.readFile(path, 'utf8')).toBe(raw); -}); - -it.each([7, 6])( - 'leaves opaque format %s data untouched without fallback until explicit preparation', - async (format) => { - await publish(older); - const previous = await selected(); - await publish(newer); - const latest = await selected(); - const path = join(latest.directory, 'overview.json'); - const accountPath = join(latest.directory, 'transactions', `${latest.accounts[0].id}.json`); - const accountBytes = await fs.readFile(accountPath, 'utf8'); - const raw = JSON.stringify({ format, overview: 'opaque old payload', accounts: null }); - await fs.writeFile(path, raw); - expect(await readReportPublication(snapshots)).toBeNull(); - expect(await readOverview(snapshots)).toBeNull(); - expect(await fs.readFile(path, 'utf8')).toBe(raw); - expect(await fs.readFile(accountPath, 'utf8')).toBe(accountBytes); - expect((await readAccountTransactions(previous, previous.accounts[0].id)).transactions).toEqual( - rows() - ); - await publish({ id: 'explicit', createdAt: new Date('2026-05-03T00:00:00Z') }); - expect(await readOverview(snapshots)).toEqual(fixture()); - expect(await fs.readFile(path, 'utf8')).toBe(raw); - expect(await fs.readFile(accountPath, 'utf8')).toBe(accountBytes); - expect(await fs.readdir(join(snapshots, '.sets'))).toHaveLength(3); - } -); - -it('leaves obsolete income-expenses and transactions namespaces untouched', async () => { - const directories = ['income-expenses', 'transactions']; - for (const name of directories) { - await fs.mkdir(join(snapshots, name), { recursive: true }); - await fs.writeFile(join(snapshots, name, 'opaque.json'), 'untouched invalid legacy data'); - } - expect(await readOverview(snapshots)).toBeNull(); - await publish(); - for (const name of directories) - expect(await fs.readFile(join(snapshots, name, 'opaque.json'), 'utf8')).toBe( - 'untouched invalid legacy data' - ); -}); - -it('rejects current overview file symlinks and non-files', async () => { - await publish(); - const path = join((await selected()).directory, 'overview.json'); - const saved = join(root, 'saved'); - await fs.rename(path, saved); - await fs.symlink(saved, path); - await expect(readOverview(snapshots)).rejects.toThrow('Unable to read overview'); - await fs.unlink(path); - await fs.mkdir(path); - await expect(readOverview(snapshots)).rejects.toThrow('Unable to read overview'); -}); - -it.each([ - { format: currentReportFormat, overview: fixture(), accounts: [] }, - { format: currentReportFormat, journal: null, overview: fixture(), accounts: [] }, - { format: currentReportFormat, journal: ' ', overview: fixture(), accounts: [] }, - { format: currentReportFormat, journal, overview: fixture(), accounts: null }, - { - format: currentReportFormat, - journal, - overview: fixture(), - accounts: [{ ...ready(), id: '../bad' }] - }, - { - format: currentReportFormat, - journal, - overview: fixture(), - accounts: [ - { ...ready(), id: '00000000' }, - { ...ready('b'), id: '00000000' } - ] - } -])('fails closed on malformed current publication metadata (%j)', async (record) => { - await publish(); - const publication = await selected(); - await fs.writeFile(join(publication.directory, 'overview.json'), JSON.stringify(record)); - await expect(readOverview(snapshots)).rejects.toSatisfy(safeError('Unable to read overview')); -}); - -it('rejects a current publication missing its required account summary', async () => { - await publish(); - const publication = await selected(); - const path = join(publication.directory, 'overview.json'); - const raw = JSON.stringify({ - format: currentReportFormat, - journal, - overview: fixture(), - accounts: [{ ...publication.accounts[0], summary: undefined }] - }); - await fs.writeFile(path, raw); - await expect(readReportPublication(snapshots)).rejects.toSatisfy( - safeError('Unable to read overview') - ); - expect(await fs.readFile(path, 'utf8')).toBe(raw); -}); - -it('rejects an undated ready summary against a dated overview, preserving the old set', async () => { - await publish(); - const pinned = await selected(); - const stage = await createReportStage(snapshots, newer); - // Same complete row setup as the happy publication; only the summary differs. - await stage.writeRows('assets:cash', rows()); - const account: PreparedAccountSummary = { ...ready(), summary: { status: 'undated' } }; - await expect(stage.publish(fixture(), [account], journal)).rejects.toSatisfy( - safeError('Unable to prepare reports') - ); - expect(await selected()).toEqual(pinned); - expect((await readAccountTransactions(pinned, pinned.accounts[0].id)).transactions).toEqual( - rows() - ); - expect(await fs.readdir(join(snapshots, '.sets'))).toEqual([basename(pinned.directory)]); -}); - -it('rejects ambiguous access with a ready summary, preserving the old set', async () => { - await publish(); - const pinned = await selected(); - const stage = await createReportStage(snapshots, newer); - const account: PreparedAccountSummary = { ...ambiguous(), summary: ready().summary }; - await expect(stage.publish(fixture(), [account], journal)).rejects.toSatisfy( - safeError('Unable to prepare reports') - ); - expect(await selected()).toEqual(pinned); - expect(await fs.readdir(join(snapshots, '.sets'))).toEqual([basename(pinned.directory)]); -}); - -it('rejects a selected current overview missing range 12 without fallback or changing bytes', async () => { - await publish(older); - const previous = await selected(); - await publish(newer); - const publication = await selected(); - const overview = { ...fixture(), ranges: { 1: range(1), 3: range(3), 6: range(6) } }; - const path = join(publication.directory, 'overview.json'); - const raw = JSON.stringify({ - format: currentReportFormat, - journal, - overview, - accounts: publication.accounts - }); - await fs.writeFile(path, raw); - await expect(readOverview(snapshots)).rejects.toSatisfy(safeError('Unable to read overview')); - expect(await fs.readFile(path, 'utf8')).toBe(raw); - expect((await readAccountTransactions(previous, previous.accounts[0].id)).transactions).toEqual( - rows() - ); -}); - -it.each([ - { label: 'missing', transactionCount: undefined }, - { label: 'negative', transactionCount: -1 }, - { label: 'unsafe', transactionCount: Number.MAX_SAFE_INTEGER + 1 } -])( - 'rejects a $label period transaction count on writes and reads without repair', - async ({ transactionCount }) => { - await publish(); - const pinned = await selected(); - const overview = fixture(); - overview.ranges![3].transactionCount = transactionCount as number; - const stage = await createReportStage(snapshots, newer, pinned.accounts); - await stage.writeRows('assets:cash', rows()); - await expect(stage.publish(overview, [ready()], journal)).rejects.toSatisfy( - safeError('Unable to prepare reports') - ); - expect(await selected()).toEqual(pinned); - expect(await fs.readdir(join(snapshots, '.sets'))).toEqual([basename(pinned.directory)]); - - const path = join(pinned.directory, 'overview.json'); - const raw = JSON.stringify({ - format: currentReportFormat, - journal, - overview, - accounts: pinned.accounts - }); - await fs.writeFile(path, raw); - await expect(readOverview(snapshots)).rejects.toSatisfy(safeError('Unable to read overview')); - expect(await fs.readFile(path, 'utf8')).toBe(raw); - expect((await readAccountTransactions(pinned, pinned.accounts[0].id)).transactions).toEqual( - rows() - ); - } -); - -it('rejects a non-string producer version on writes and reads without repair', async () => { - await publish(); - const pinned = await selected(); - const overview = { ...fixture(), hledgerVersion: 19995 } as unknown as OverviewData; - const stage = await createReportStage(snapshots, newer, pinned.accounts); - await stage.writeRows('assets:cash', rows()); - await expect(stage.publish(overview, [ready()], journal)).rejects.toSatisfy( - safeError('Unable to prepare reports') - ); - expect(await selected()).toEqual(pinned); - expect(await fs.readdir(join(snapshots, '.sets'))).toEqual([basename(pinned.directory)]); - - const path = join(pinned.directory, 'overview.json'); - const raw = JSON.stringify({ - format: currentReportFormat, - journal, - overview, - accounts: pinned.accounts - }); - await fs.writeFile(path, raw); - await expect(readOverview(snapshots)).rejects.toSatisfy(safeError('Unable to read overview')); - expect(await fs.readFile(path, 'utf8')).toBe(raw); -}); - -it('rejects publication without range 12, preserving the previous readable pinned set', async () => { - await publish(); - const pinned = await selected(); - const stage = await createReportStage(snapshots, newer); - await stage.writeRows('assets:cash', rows()); - const overview = { ...fixture(), ranges: { 1: range(1), 3: range(3), 6: range(6) } }; - await expect(stage.publish(overview as OverviewData, [ready()], journal)).rejects.toSatisfy( - safeError('Unable to prepare reports') - ); - expect(await selected()).toEqual(pinned); - expect((await readAccountTransactions(pinned, pinned.accounts[0].id)).transactions).toEqual( - rows() - ); - expect(await fs.readdir(join(snapshots, '.sets'))).toEqual([basename(pinned.directory)]); -}); - -it.each(['not JSON: private text', 'null', '[]'])( - 'rejects malformed selected JSON without fallback (%s)', - async (raw) => { - await publish(older); - await publish(newer); - const path = join((await selected()).directory, 'overview.json'); - await fs.writeFile(path, raw); - await expect(readOverview(snapshots)).rejects.toSatisfy(safeError('Unable to read overview')); - expect(await fs.readFile(path, 'utf8')).toBe(raw); - } -); - -it('never opens or replaces obsolete observation files', async () => { - await fs.mkdir(join(snapshots, 'dashboard'), { recursive: true }); - const paths = [join(snapshots, 'source.json'), join(snapshots, 'dashboard', 'inputs.json')]; - for (const path of paths) await fs.writeFile(path, 'untouched invalid legacy metadata'); - const open = fs.open; - vi.spyOn(fs, 'open').mockImplementation(async (path, ...args) => { - expect(paths).not.toContain(path); - return open(path, ...args); - }); - expect(await readReportPublication(snapshots)).toBeNull(); - await publish(); - expect((await selected()).journal).toBe(journal); - for (const path of paths) - expect(await fs.readFile(path, 'utf8')).toBe('untouched invalid legacy metadata'); -}); - -it('publishes without reading selected financial files and keeps committed sets after alias replacement fails', async () => { - await publish(newer); - const pinned = await selected(); - const open = fs.open; - vi.spyOn(fs, 'open').mockImplementation(async (path, flags, ...args) => { - if (String(path).endsWith('.json')) expect(flags).toBeTypeOf('number'); - if (String(path).endsWith('.json')) expect(Number(flags) & 1).toBe(1); - return open(path, flags, ...args); - }); - expect(await publish(older)).toBeUndefined(); - const stage = await createReportStage(snapshots, newer); + await stage.writeRows('assets:cash', rows(1, 'committed')); const rename = fs.rename; vi.spyOn(fs, 'rename').mockImplementation(async (source, destination) => { if (destination === join(snapshots, 'current')) throw new Error('private alias failure'); await rename(source, destination); }); - await expect(stage.publish(fixture(), [], journal)).rejects.toSatisfy( - safeError('Unable to prepare reports') - ); + await expect( + stage.publish({ ...fixture(), preparedAt: 'committed' }, [ready()], journal) + ).rejects.toSatisfy(safeError('Unable to prepare reports')); await stage.discard(); + expect(reader).not.toHaveBeenCalled(); vi.restoreAllMocks(); - expect(await fs.readdir(join(snapshots, '.sets'))).toHaveLength(3); + const publication = await selected(); + expect(publication.overview.preparedAt).toBe('committed'); + expect( + (await readAccountTransactions(publication, publication.accounts[0].id)).transactions + ).toEqual(rows(1, 'committed')); + expect(await fs.readdir(join(snapshots, '.sets'))).toHaveLength(2); expect(await fs.readlink(join(snapshots, 'current'))).toBe(`.sets/${basename(pinned.directory)}`); - expect((await readAccountTransactions(pinned, pinned.accounts[0].id)).transactions).toEqual( +}); + +it('closes read-only connections on success and on catalog or overview validation failure', async () => { + await publish(); + const openDatabase = databases.openReadOnlyReportDatabase; + const connectionClosures: ReturnType[] = []; + vi.spyOn(databases, 'openReadOnlyReportDatabase').mockImplementation(async (path) => { + const database = await openDatabase(path); + if (database) { + const close = vi.fn(); + connectionClosures.push(close); + const destroy = database.destroy.bind(database); + vi.spyOn(database, 'destroy').mockImplementation(async () => { + await destroy(); + close(); + }); + } + return database; + }); + const publication = await selected(); + expect(connectionClosures.shift()).toHaveBeenCalledOnce(); + const account = publication.accounts[0]; + expect((await readAccountTransactions(publication, account.id)).transactions).toEqual(rows()); + expect(connectionClosures.shift()).toHaveBeenCalledOnce(); + const mismatched = { + ...publication, + accounts: [{ ...account, name: 'not the persisted account' }] + }; + await expect(readAccountTransactions(mismatched, account.id)).rejects.toSatisfy( + safeError('Unable to read account transactions') + ); + expect(connectionClosures.shift()).toHaveBeenCalledOnce(); + const overview = fixture(); + const incomplete = { + ...overview, + ranges: { 1: overview.ranges![1], 3: overview.ranges![3], 6: overview.ranges![6] } + }; + const database = new DatabaseSync(join(publication.directory, 'reports.sqlite')); + try { + database.prepare('UPDATE publication SET overview_json = ?').run(JSON.stringify(incomplete)); + } finally { + database.close(); + } + await expect(readReportPublication(snapshots)).rejects.toSatisfy( + safeError('Unable to read overview') + ); + expect(connectionClosures.shift()).toHaveBeenCalledOnce(); +}); + +it('rejects unsafe and uncatalogued account IDs before opening financial data', async () => { + const publication: ReportPublication = { + overview: fixture(), + journal, + accounts: [{ id: 'cash0001', ...ready() }], + directory: join(snapshots, '.sets', 'must-not-be-opened') + }; + const reader = vi.spyOn(databases, 'openReadOnlyReportDatabase'); + await expect(readAccountTransactions(publication, '../escape')).rejects.toSatisfy( + safeError('Unable to read account transactions') + ); + await expect(readAccountTransactions(publication, '00000000')).rejects.toSatisfy( + safeError('Unable to read account transactions') + ); + expect(reader).not.toHaveBeenCalled(); +}); + +it('reports a bad selected database without falling back or changing its bytes', async () => { + await publish(); + const previous = await selected(); + await publish(newer); + const path = join((await selected()).directory, 'reports.sqlite'); + const bytes = Buffer.from('not SQLite: fictional private report data'); + await fs.writeFile(path, bytes); + await expect(readOverview(snapshots)).rejects.toSatisfy(safeError('Unable to read overview')); + expect(await fs.readFile(path)).toEqual(bytes); + expect((await readAccountTransactions(previous, previous.accounts[0].id)).transactions).toEqual( rows() ); }); -it.each([undefined, ' '])( - 'rejects missing journal identity on new writes (%j)', - async (journal) => { - await publish(); - const pinned = await selected(); - const stage = await createReportStage(snapshots, newer); - await expect(stage.publish(fixture(), [], journal as string)).rejects.toSatisfy( - safeError('Unable to prepare reports') - ); - expect(await selected()).toEqual(pinned); - expect(await fs.readdir(join(snapshots, '.sets'))).toEqual([basename(pinned.directory)]); +it('withholds an unsupported selected SQLite version without migration or fallback', async () => { + await publish(); + const previous = await selected(); + await publish(newer); + const path = join((await selected()).directory, 'reports.sqlite'); + const database = new DatabaseSync(path); + try { + database.exec('PRAGMA user_version = 999'); + } finally { + database.close(); } -); + const bytes = await fs.readFile(path); + expect(await readReportPublication(snapshots)).toBeNull(); + expect(await readOverview(snapshots)).toBeNull(); + expect(await fs.readFile(path)).toEqual(bytes); + expect((await readAccountTransactions(previous, previous.accounts[0].id)).transactions).toEqual( + rows() + ); +}); + +it('leaves a selected JSON-only generation unavailable and untouched until explicit preparation', async () => { + await publish(); + const previous = await selected(); + const token = `${String(newer.createdAt.getTime()).padStart(16, '0')}-${newer.id}-00000000-0000-4000-8000-000000000000`; + const directory = join(snapshots, '.sets', token); + await fs.mkdir(join(directory, 'transactions'), { recursive: true, mode: 0o700 }); + const overviewPath = join(directory, 'overview.json'); + const accountPath = join(directory, 'transactions', `${previous.accounts[0].id}.json`); + const overviewBytes = JSON.stringify({ + format: 8, + journal, + overview: fixture(), + accounts: previous.accounts + }); + const accountBytes = 'opaque legacy account payload'; + await fs.writeFile(overviewPath, overviewBytes); + await fs.writeFile(accountPath, accountBytes); + expect(await readReportPublication(snapshots)).toBeNull(); + expect(await readOverview(snapshots)).toBeNull(); + expect((await readAccountTransactions(previous, previous.accounts[0].id)).transactions).toEqual( + rows() + ); + await publish({ id: 'explicit', createdAt: new Date(newer.createdAt.getTime() + 1) }); + expect(await readOverview(snapshots)).toEqual(fixture()); + expect(await fs.readFile(overviewPath, 'utf8')).toBe(overviewBytes); + expect(await fs.readFile(accountPath, 'utf8')).toBe(accountBytes); + expect((await fs.readdir(directory)).sort()).toEqual(['overview.json', 'transactions']); +}); + +it('rejects a symlink in place of the selected completed directory', async () => { + await publish(); + const publication = await selected(); + const saved = join(root, 'saved'); + await fs.rename(publication.directory, saved); + await fs.symlink(saved, publication.directory); + await expect(readOverview(snapshots)).rejects.toSatisfy(safeError('Unable to read overview')); + await expect(readAccountTransactions(publication, publication.accounts[0].id)).rejects.toSatisfy( + safeError('Unable to read account transactions') + ); +}); + +it('requires a journal identity before replacing a publication', async () => { + await publish(); + const pinned = await selected(); + const stage = await createReportStage(snapshots, newer); + await expect(stage.publish(fixture(), [], ' ')).rejects.toSatisfy( + safeError('Unable to prepare reports') + ); + expect(await selected()).toEqual(pinned); + expect(await fs.readdir(join(snapshots, '.sets'))).toEqual([basename(pinned.directory)]); +}); + +it('rejects a path-like request ID before creating storage', async () => { + await expect(createReportStage(snapshots, { ...older, id: '../escape' })).rejects.toSatisfy( + safeError('Unable to prepare reports') + ); + expect(await fs.readdir(root)).toEqual([]); +}); diff --git a/packages/snapshots/src/reports/publication.ts b/packages/snapshots/src/reports/publication.ts index f528eff..1c8ad2e 100644 --- a/packages/snapshots/src/reports/publication.ts +++ b/packages/snapshots/src/reports/publication.ts @@ -3,30 +3,36 @@ import type { PreparedAccountSummary, PreparedAccountRow } from '@hledger-reports/hledger'; -import { randomInt, randomUUID } from 'node:crypto'; -import { constants } from 'node:fs'; -import { mkdir, open, rename, rm, symlink, unlink } from 'node:fs/promises'; -import type { FileHandle } from 'node:fs/promises'; -import { basename, dirname, join } from 'node:path'; -import { namesIn, privateDirectory, readJson, writeJson } from '../files.js'; +import { + createReportDatabase, + openReadOnlyReportDatabase, + runMigrations, + type ReportConnection, + type ReportTransaction +} from '@hledger-reports/db'; import { hasOverviewContainers, - hasAccountRowContainers, - hasStoredReportData, isAccountCatalog, isAccountId, isAccountSummary, - isObject, + readAccountRows, + readReportData, storedSummary, summaryMatchesDate, + writeAccountRows, + writeReportData, type StoredAccount, type StoredReportData } from '@hledger-reports/db/reports'; +import { randomInt, randomUUID } from 'node:crypto'; +import { lstat, mkdir, rename, rm, symlink, unlink } from 'node:fs/promises'; +import { basename, dirname, join } from 'node:path'; +import { namesIn, privateDirectory } from '../files.js'; const runIdPattern = /^[A-Za-z0-9_-]{1,128}(?![\s\S])/; const setPattern = /^(\d{16}-[A-Za-z0-9_-]{1,128})-[0-9a-f]{8}-[0-9a-f]{4}-4[0-9a-f]{3}-[89ab][0-9a-f]{3}-[0-9a-f]{12}(?![\s\S])/; -const currentReportFormat = 8; +const databaseName = 'reports.sqlite'; export interface ReportPublication extends StoredReportData { /** Private, pinned completed-set directory; never expose this in financial data. */ @@ -59,6 +65,17 @@ async function latestSet(snapshotDirectory: string): Promise .at(-1); } +async function openSelectedDatabase(directory: string): Promise { + try { + return await openReadOnlyReportDatabase(join(directory, databaseName)); + } catch (error) { + if ((error as NodeJS.ErrnoException).code !== 'ENOENT') throw error; + const legacy = await lstat(join(directory, 'overview.json')); + if (!legacy.isFile()) throw error; + return null; + } +} + export async function readReportPublication( snapshotDirectory: string ): Promise { @@ -67,17 +84,14 @@ export async function readReportPublication( if (!latest) return null; const directory = join(snapshotDirectory, '.sets', latest); await privateDirectory(directory); - const record = await readJson(directory, 'overview.json'); - if (!isObject(record)) throw new Error(); - if (record.format !== currentReportFormat) return null; - if (!hasStoredReportData(record)) throw new Error(); - return { - // Financial rows remain producer-owned after checking current header/containers. - overview: record.overview, - journal: record.journal, - accounts: record.accounts, - directory - }; + const database = await openSelectedDatabase(directory); + if (!database) return null; + try { + const data = await readReportData(database); + return { ...data, directory }; + } finally { + await database.destroy(); + } } catch { throw new Error('Unable to read overview'); } @@ -100,23 +114,17 @@ export async function readAccountTransactions( throw new Error(); await privateDirectory(dirname(directory)); await privateDirectory(directory); - const transactionsDirectory = join(directory, 'transactions'); - await privateDirectory(transactionsDirectory); - const record = await readJson(transactionsDirectory, `${id}.json`); - if (!isObject(record) || record.id !== id || record.name !== account.name) throw new Error(); - if (account.access === 'ambiguous') { - if (record.transactions !== null) throw new Error(); - } else if ( - !Array.isArray(record.transactions) || - record.transactions.length !== account.rowCount || - !record.transactions.every(hasAccountRowContainers) - ) - throw new Error(); - return { - id, - name: account.name, - transactions: record.transactions as PreparedAccountRow[] | null - }; + const database = await openReadOnlyReportDatabase(join(directory, databaseName)); + if (!database) throw new Error(); + try { + return { + id, + name: account.name, + transactions: await readAccountRows(database, account) + }; + } finally { + await database.destroy(); + } } catch { throw new Error('Unable to read account transactions'); } @@ -150,6 +158,34 @@ export async function createReportStage( previousAccounts?: readonly StoredAccount[] | null ): Promise { let staging: string | undefined; + let database: ReportConnection | undefined; + let transaction: ReportTransaction | undefined; + let databaseClosed = false; + let state: 'active' | 'discarding' | 'completed' | 'discarded' = 'active'; + + async function closeDatabase(): Promise { + if (databaseClosed) return; + await database?.destroy(); + databaseClosed = true; + } + + async function discard(): Promise { + if (state === 'completed' || state === 'discarded') return; + state = 'discarding'; + let rollbackFailed = false; + if (!databaseClosed && transaction && !transaction.isCommitted && !transaction.isRolledBack) { + try { + await transaction.rollback().execute(); + } catch { + rollbackFailed = true; + } + } + await closeDatabase(); + if (staging) await rm(staging, { recursive: true, force: true }); + state = 'discarded'; + if (rollbackFailed) throw new Error('Unable to prepare reports'); + } + try { const timestamp = String(run.createdAt.getTime()).padStart(16, '0'); if (!runIdPattern.test(run.id) || !/^\d{16}$/.test(timestamp)) throw new Error(); @@ -177,31 +213,12 @@ export async function createReportStage( const temporary = join(sets, `.${uuid}.tmp`); await mkdir(temporary, { mode: 0o700 }); staging = temporary; - const transactionsDirectory = join(temporary, 'transactions'); - await privateDirectory(transactionsDirectory, true); - let state: 'active' | 'completed' | 'discarded' = 'active'; + database = await createReportDatabase(join(temporary, databaseName)); + await runMigrations(database); + transaction = await database.startTransaction().execute(); + const writer = transaction; const written = new Map(); - let current: { name: string; file: FileHandle } | undefined; - - async function closeCurrent(): Promise { - if (!current) return; - const { file } = current; - current = undefined; - try { - await file.writeFile(']}', 'utf8'); - } finally { - await file.close(); - } - } - async function discard(): Promise { - if (state !== 'active') return; - state = 'discarded'; - try { - await closeCurrent(); - } finally { - await rm(temporary, { recursive: true, force: true }); - } - } + let currentName: string | undefined; async function operation(action: () => Promise): Promise { try { if (state !== 'active') throw new Error(); @@ -215,24 +232,13 @@ export async function createReportStage( writeRows(name, rows) { return operation(async () => { if (typeof name !== 'string' || !Array.isArray(rows)) throw new Error(); - if (current?.name !== name) { - await closeCurrent(); + if (currentName !== name) { if (written.has(name)) throw new Error(); - const id = accountId(name); - const header = `{"id":${JSON.stringify(id)},"name":${JSON.stringify(name)},"transactions":[`; - const file = await open( - join(transactionsDirectory, `${id}.json`), - constants.O_WRONLY | constants.O_CREAT | constants.O_EXCL | constants.O_NOFOLLOW, - 0o600 - ); - current = { name, file }; - written.set(name, { id, count: 0 }); - await file.writeFile(header, 'utf8'); + currentName = name; + written.set(name, { id: accountId(name), count: 0 }); } const metadata = written.get(name)!; - if (!rows.length) return; - const json = (metadata.count ? ',' : '') + JSON.stringify(rows).slice(1, -1); - await current!.file.writeFile(json, 'utf8'); + await writeAccountRows(writer, name, metadata.count, rows); metadata.count += rows.length; }); }, @@ -245,7 +251,6 @@ export async function createReportStage( !Array.isArray(accounts) ) throw new Error(); - await closeCurrent(); const remaining = new Set(written.keys()); const names = new Set(); const catalog: StoredAccount[] = []; @@ -264,17 +269,8 @@ export async function createReportStage( } else if ((metadata?.count ?? 0) !== account.rowCount) { throw new Error(); } - const id = metadata?.id ?? accountId(account.name); - if (!metadata) { - const transactions = account.access === 'ambiguous' ? null : []; - await writeJson(transactionsDirectory, `${id}.json`, { - id, - name: account.name, - transactions - }); - } catalog.push({ - id, + id: metadata?.id ?? accountId(account.name), name: account.name, access: account.access, rowCount: account.rowCount, @@ -282,13 +278,9 @@ export async function createReportStage( }); } if (remaining.size) throw new Error(); - await writeJson(temporary, 'overview.json', { - format: currentReportFormat, - journal, - overview, - accounts: catalog - }); - // The directory rename is the one commit point for the entire report set. + await writeReportData(writer, { overview, accounts: catalog, journal }); + await writer.commit().execute(); + await closeDatabase(); await rename(temporary, join(sets, token)); state = 'completed'; await updateCurrent(snapshotDirectory); @@ -303,7 +295,7 @@ export async function createReportStage( } }; } catch { - if (staging) await rm(staging, { recursive: true, force: true }).catch(() => {}); + await discard().catch(() => {}); throw new Error('Unable to prepare reports'); } } -- 2.54.0