tests/unit/ledger/compaction.test.ts

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<string> {
	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<Ledger> {
		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<void> {
			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<void>((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<void>((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');
	});
});