import { mkdtemp, rm } from 'node:fs/promises'; import { tmpdir } from 'node:os'; import { join } from 'node:path'; import { afterEach, beforeEach, describe, expect, it, vi } from 'vitest'; import type { ExpenseBody } from '../../../src/lib/domain/types'; import { parseJsonl } from '../../../src/lib/server/ledger/events'; import { Ledger } from '../../../src/lib/server/ledger/ledger'; import { LocalFsStore } from '../../../src/lib/server/storage/local-fs-store'; import { decodeText, encodeText } from '../../../src/lib/server/storage/store'; import type { BucketStore } from '../../../src/lib/server/storage/store'; import { FailingStore, type Fault } from '../helpers/failing-store'; const body = (description: string): ExpenseBody => ({ description, amountCents: 300, currency: 'USD', date: '2026-08-01', paidBy: 'u_a', split: { mode: 'equal', participants: ['u_a', 'u_b', 'u_c'] }, shares: { u_a: 100, u_b: 100, u_c: 100 } }); /** Distinct event ids on disk, tolerating the duplicate lines a lost-ack retry is allowed to leave. */ function distinctIds(text: string): Set { return new Set(text.trim().split('\n').map((l) => JSON.parse(l).id as string)); } describe('Ledger.compact', () => { let dir: string; let base: LocalFsStore; let day = Date.UTC(2026, 7, 29, 12); const clock = () => new Date(day); beforeEach(async () => { dir = await mkdtemp(join(tmpdir(), 'splitwise-compact-')); base = new LocalFsStore(dir); }); afterEach(() => rm(dir, { recursive: true, force: true })); async function seed(n: number): Promise { const ledger = await Ledger.load(base, { clock }); for (let i = 0; i < n; i++) await ledger.createExpense('u_a', body(`e${i}`)); return ledger; } it('does nothing when there are no events', async () => { const ledger = await Ledger.load(base, { clock }); expect(await ledger.compact()).toEqual({ appended: 0, deleted: 0, backedUp: false }); expect(await base.get('ledger.jsonl')).toBeNull(); }); it('moves events into the ledger in ULID order and deletes the files', async () => { const ledger = await seed(3); const result = await ledger.compact(); expect(result).toEqual({ appended: 3, deleted: 3, backedUp: false }); expect(await base.list('events/')).toEqual([]); const lines = decodeText((await base.get('ledger.jsonl'))!).trim().split('\n'); expect(lines).toHaveLength(3); expect(lines.map((l) => JSON.parse(l).data.description)).toEqual(['e0', 'e1', 'e2']); expect(ledger.stats()).toEqual({ pendingEvents: 0, ledgerEvents: 3, ledgerExists: true }); const reloaded = await Ledger.load(base, { clock }); expect(reloaded.listExpenses().map((e) => e.description).sort()).toEqual(['e0', 'e1', 'e2']); }); it('appends on the second compaction and backs up once per day', async () => { const ledger = await seed(2); await ledger.compact(); await ledger.createExpense('u_a', body('later')); const second = await ledger.compact(); expect(second).toEqual({ appended: 1, deleted: 1, backedUp: true }); const backup = decodeText((await base.get('backups/ledger-2026-08-29.jsonl'))!).trim().split('\n'); expect(backup).toHaveLength(2); expect(decodeText((await base.get('ledger.jsonl'))!).trim().split('\n')).toHaveLength(3); await ledger.createExpense('u_a', body('same day')); expect((await ledger.compact()).backedUp).toBe(false); day += 24 * 3600 * 1000; await ledger.createExpense('u_a', body('next day')); expect((await ledger.compact()).backedUp).toBe(true); expect(await base.get('backups/ledger-2026-08-30.jsonl')).not.toBeNull(); }); const faults: Fault[] = [ { method: 'copy', when: 'before' }, { method: 'copy', when: 'after' }, { method: 'append', when: 'before' }, { method: 'append', when: 'after' }, { method: 'delete', when: 'before' }, { method: 'delete', when: 'after' } ]; for (const fault of faults) { it(`loses nothing when the store fails ${fault.when} ${fault.method}`, async () => { // First compaction creates the ledger so later ones exercise copy+append. const seeded = await seed(2); await seeded.compact(); await seeded.createExpense('u_a', body('x')); await seeded.createExpense('u_a', body('y')); const failing = new FailingStore(base, fault); const ledger = await Ledger.load(failing, { clock }); await expect(ledger.compact()).rejects.toThrow(`injected failure ${fault.when} ${fault.method}`); // Simulate a restart from whatever the store now contains right after the crash. const after = await Ledger.load(base, { clock }); expect(after.listExpenses().map((e) => e.description).sort()).toEqual(['e0', 'e1', 'x', 'y']); // The SAME instance, retried without a restart, must also fully self-heal. A premature // update to in-memory bookkeeping (e.g. marking an event as ledgered before its write is // durable) would be invisible to a fresh reload but would make this retry silently drop // the event once its file is deleted. await ledger.compact(); expect(await base.list('events/')).toEqual([]); expect(distinctIds(decodeText((await base.get('ledger.jsonl'))!)).size).toBe(4); expect(ledger.listExpenses().map((e) => e.description).sort()).toEqual(['e0', 'e1', 'x', 'y']); expect(ledger.stats().pendingEvents).toBe(0); // A retry from a fresh restart also finishes the job. const retried = await Ledger.load(base, { clock }); await retried.compact(); expect(await base.list('events/')).toEqual([]); // Duplicated lines are allowed on disk (lost ack) but never lost, and fold dedups them. expect(distinctIds(decodeText((await base.get('ledger.jsonl'))!)).size).toBe(4); const final = await Ledger.load(base, { clock }); expect(final.listExpenses().map((e) => e.description).sort()).toEqual(['e0', 'e1', 'x', 'y']); expect(final.stats().pendingEvents).toBe(0); }); } describe('heals a truncated ledger even when the rewrite crashes', () => { /** * Compacts 2 events, then leaves a 3rd fully-written pending event file plus a truncated * trailing line appended to ledger.jsonl, mirroring an append that crashed mid-line. */ async function seedTruncated(): Promise { const seeded = await seed(2); await seeded.compact(); await seeded.createExpense('u_a', body('third')); const [pendingFile] = await base.list('events/'); const partialLine = decodeText((await base.get(pendingFile.path))!).slice(0, 24); const ledgerText = decodeText((await base.get('ledger.jsonl'))!); await base.put('ledger.jsonl', encodeText(ledgerText + partialLine)); } const putFaults: Fault[] = [ { method: 'put', when: 'before' }, { method: 'put', when: 'after' } ]; for (const fault of putFaults) { it(`heals the truncated ledger when the rewrite fails ${fault.when} ${fault.method}`, async () => { await seedTruncated(); const failing = new FailingStore(base, fault); const ledger = await Ledger.load(failing, { clock }); await expect(ledger.compact()).rejects.toThrow(`injected failure ${fault.when} ${fault.method}`); // Right after the crash, a fresh reload must already see every event. const after = await Ledger.load(base, { clock }); expect(after.listExpenses().map((e) => e.description).sort()).toEqual(['e0', 'e1', 'third']); // The SAME instance, retried without a restart, converges to a fully parseable ledger. await ledger.compact(); expect(await base.list('events/')).toEqual([]); const parsed = parseJsonl(decodeText((await base.get('ledger.jsonl'))!)); expect(parsed.truncated).toBe(false); expect(new Set(parsed.events.map((e) => e.id)).size).toBe(3); expect(ledger.listExpenses().map((e) => e.description).sort()).toEqual(['e0', 'e1', 'third']); expect(ledger.stats().pendingEvents).toBe(0); // A restart-based retry converges too. const retried = await Ledger.load(base, { clock }); await retried.compact(); expect(await base.list('events/')).toEqual([]); const reparsed = parseJsonl(decodeText((await base.get('ledger.jsonl'))!)); expect(reparsed.truncated).toBe(false); const final = await Ledger.load(base, { clock }); expect(final.listExpenses().map((e) => e.description).sort()).toEqual(['e0', 'e1', 'third']); expect(final.stats().pendingEvents).toBe(0); }); } }); it('writes made while a compaction is in flight are not lost', async () => { const setup = await seed(1); await setup.compact(); await setup.createExpense('u_a', body('toCompact')); let appendCalled = false; let releaseAppend!: () => void; const blocked = new Promise((resolve) => { releaseAppend = resolve; }); // A store that behaves normally except append blocks until released, so we can catch // compaction with its write in flight and prove a concurrent write is queued behind it. const blocking: BucketStore = { get: (p) => base.get(p), put: (p, b, ct) => base.put(p, b, ct), append: async (p, b) => { appendCalled = true; await blocked; await base.append(p, b); }, list: (p) => base.list(p), delete: (p) => base.delete(p), copy: (s, d) => base.copy(s, d) }; const ledger = await Ledger.load(blocking, { clock }); const compaction = ledger.compact(); await vi.waitFor(() => expect(appendCalled).toBe(true)); const write = ledger.createExpense('u_a', body('during')); let writeSettled = false; write.then( () => { writeSettled = true; }, () => { writeSettled = true; } ); await new Promise((r) => setTimeout(r, 0)); expect(writeSettled).toBe(false); releaseAppend(); await Promise.all([compaction, write]); const reloaded = await Ledger.load(base, { clock }); expect(reloaded.listExpenses().map((e) => e.description).sort()).toEqual(['during', 'e0', 'toCompact']); const pendingFiles = await base.list('events/'); expect(pendingFiles).toHaveLength(1); expect(JSON.parse(decodeText((await base.get(pendingFiles[0].path))!)).data.description).toBe('during'); }); });