feat(ledger): add Ledger service with event write path and compaction

src/lib/server/ledger/errors.ts

@@ -0,0 +1,23 @@
+export class ConflictError extends Error {
+	constructor(
+		public readonly entityId: string,
+		public readonly currentVersion: number
+	) {
+		super(`Entity ${entityId} was modified (current version ${currentVersion})`);
+		this.name = 'ConflictError';
+	}
+}
+
+export class NotFoundError extends Error {
+	constructor(public readonly entityId: string) {
+		super(`Entity ${entityId} not found`);
+		this.name = 'NotFoundError';
+	}
+}
+
+export class ReadOnlyError extends Error {
+	constructor() {
+		super('The app is in read-only mode: another instance holds the writer lock');
+		this.name = 'ReadOnlyError';
+	}
+}

src/lib/server/ledger/ledger.ts

@@ -0,0 +1,243 @@
+import { ulid as defaultUlid } from '../../domain/ulid';
+import type { Expense, ExpenseBody, Settlement, SettlementBody } from '../../domain/types';
+import { decodeText, type BucketStore } from '../storage/store';
+import { ConflictError, NotFoundError, ReadOnlyError } from './errors';
+import { decodeEvent, encodeEvent, encodeJsonl, EVENT_KEY_RE, EVENTS_PREFIX, eventKey, LEDGER_PATH, parseJsonl, type LedgerEvent } from './events';
+import { applyEvent, foldEvents, type FoldedState } from './fold';
+
+export type Clock = () => Date;
+
+export interface LedgerOptions {
+	clock?: Clock;
+	newId?: () => string;
+}
+
+export interface LedgerStats {
+	pendingEvents: number;
+	ledgerEvents: number;
+	ledgerExists: boolean;
+}
+
+type DistributiveOmit<T, K extends PropertyKey> = T extends unknown ? Omit<T, K> : never;
+/** A LedgerEvent before the Ledger assigns id, ts and actor. */
+export type NewEvent = DistributiveOmit<LedgerEvent, 'id' | 'ts' | 'actor'>;
+
+export interface CompactionResult {
+	/** Events appended to the ledger. */
+	appended: number;
+	/** Event files deleted. */
+	deleted: number;
+	backedUp: boolean;
+}
+
+/**
+ * In-memory view of all expenses and settlements, backed by an append-only event log.
+ * All mutations go through a single async queue so writes never interleave.
+ */
+export class Ledger {
+	readOnly = false;
+	private queue: Promise<unknown> = Promise.resolve();
+	private lastBackupDay: string | null = null;
+
+	private constructor(
+		private readonly store: BucketStore,
+		private readonly state: FoldedState,
+		/** Ids of events already present in ledger.jsonl. */
+		private readonly ledgerIds: Set<string>,
+		private ledgerExists: boolean,
+		/** Event files in events/ that have not been compacted yet, keyed by object path. */
+		private readonly pending: Map<string, LedgerEvent>,
+		private readonly clock: Clock,
+		private readonly newId: () => string
+	) {}
+
+	static async load(store: BucketStore, opts: LedgerOptions = {}): Promise<Ledger> {
+		const clock = opts.clock ?? (() => new Date());
+		const newId = opts.newId ?? defaultUlid;
+
+		const listing = await store.list(EVENTS_PREFIX);
+		let ledgerBytes = await store.get(LEDGER_PATH);
+		const pending = new Map<string, LedgerEvent>();
+		let sawMissing = false;
+		for (const entry of listing) {
+			if (!EVENT_KEY_RE.test(entry.path)) continue;
+			const bytes = await store.get(entry.path);
+			if (!bytes) {
+				sawMissing = true;
+				continue;
+			}
+			pending.set(entry.path, decodeEvent(bytes));
+		}
+		// A compactor deleted an event after publishing a ledger that contains it: re-read the ledger.
+		if (sawMissing) ledgerBytes = await store.get(LEDGER_PATH);
+
+		const ledgerEvents = ledgerBytes ? parseJsonl(decodeText(ledgerBytes)) : [];
+		const ledgerIds = new Set(ledgerEvents.map((e) => e.id));
+		const sortedPending = [...pending.values()].sort((a, b) => (a.id < b.id ? -1 : a.id > b.id ? 1 : 0));
+		const state = foldEvents([...ledgerEvents, ...sortedPending]);
+		return new Ledger(store, state, ledgerIds, ledgerBytes !== null, pending, clock, newId);
+	}
+
+	// ---- reads -------------------------------------------------------------
+
+	getExpense(id: string): Expense | undefined {
+		return this.state.expenses.get(id);
+	}
+	listExpenses(): Expense[] {
+		return [...this.state.expenses.values()];
+	}
+	getSettlement(id: string): Settlement | undefined {
+		return this.state.settlements.get(id);
+	}
+	listSettlements(): Settlement[] {
+		return [...this.state.settlements.values()];
+	}
+	history(entityId: string): LedgerEvent[] {
+		return [...(this.state.history.get(entityId) ?? [])];
+	}
+	stats(): LedgerStats {
+		return { pendingEvents: this.pending.size, ledgerEvents: this.ledgerIds.size, ledgerExists: this.ledgerExists };
+	}
+
+	// ---- writes ------------------------------------------------------------
+
+	/** `id` may be supplied so callers can store a photo under the expense id before the event is written. */
+	createExpense(actor: string, body: ExpenseBody, id = `e_${this.newId()}`): Promise<Expense> {
+		return this.run(async () => {
+			const now = this.clock().toISOString();
+			const expense: Expense = { ...body, id, version: 1, createdBy: actor, createdAt: now, updatedBy: actor, updatedAt: now };
+			await this.writeEvent({ kind: 'expense', op: 'upsert', entityId: expense.id, version: 1, data: expense }, actor, now);
+			return expense;
+		});
+	}
+
+	updateExpense(actor: string, id: string, body: ExpenseBody, expectedVersion: number): Promise<Expense> {
+		return this.run(async () => {
+			const current = this.requireExpense(id, expectedVersion);
+			const now = this.clock().toISOString();
+			const expense: Expense = {
+				...body, id, version: current.version + 1,
+				createdBy: current.createdBy, createdAt: current.createdAt, updatedBy: actor, updatedAt: now
+			};
+			await this.writeEvent({ kind: 'expense', op: 'upsert', entityId: id, version: expense.version, data: expense }, actor, now);
+			return expense;
+		});
+	}
+
+	deleteExpense(actor: string, id: string, expectedVersion: number): Promise<void> {
+		return this.run(async () => {
+			const current = this.requireExpense(id, expectedVersion);
+			await this.writeEvent({ kind: 'expense', op: 'delete', entityId: id, version: current.version + 1 }, actor, this.clock().toISOString());
+		});
+	}
+
+	createSettlement(actor: string, body: SettlementBody): Promise<Settlement> {
+		return this.run(async () => {
+			const now = this.clock().toISOString();
+			const settlement: Settlement = { ...body, id: `s_${this.newId()}`, version: 1, createdBy: actor, createdAt: now, updatedBy: actor, updatedAt: now };
+			await this.writeEvent({ kind: 'settlement', op: 'upsert', entityId: settlement.id, version: 1, data: settlement }, actor, now);
+			return settlement;
+		});
+	}
+
+	updateSettlement(actor: string, id: string, body: SettlementBody, expectedVersion: number): Promise<Settlement> {
+		return this.run(async () => {
+			const current = this.requireSettlement(id, expectedVersion);
+			const now = this.clock().toISOString();
+			const settlement: Settlement = {
+				...body, id, version: current.version + 1,
+				createdBy: current.createdBy, createdAt: current.createdAt, updatedBy: actor, updatedAt: now
+			};
+			await this.writeEvent({ kind: 'settlement', op: 'upsert', entityId: id, version: settlement.version, data: settlement }, actor, now);
+			return settlement;
+		});
+	}
+
+	deleteSettlement(actor: string, id: string, expectedVersion: number): Promise<void> {
+		return this.run(async () => {
+			const current = this.requireSettlement(id, expectedVersion);
+			await this.writeEvent({ kind: 'settlement', op: 'delete', entityId: id, version: current.version + 1 }, actor, this.clock().toISOString());
+		});
+	}
+
+	// ---- compaction --------------------------------------------------------
+
+	/**
+	 * Appends every pending event to ledger.jsonl, then deletes the event files.
+	 * Safe to crash at any point: nothing acknowledged is ever lost, and duplicates are removed by the fold.
+	 */
+	compact(): Promise<CompactionResult> {
+		return this.run(async () => {
+			if (this.readOnly) throw new ReadOnlyError();
+			const listing = await this.store.list(EVENTS_PREFIX);
+			const keys = listing.map((e) => e.path).filter((p) => EVENT_KEY_RE.test(p)).sort();
+
+			const toAppend: LedgerEvent[] = [];
+			const toDelete: string[] = [];
+			for (const key of keys) {
+				let event = this.pending.get(key);
+				if (!event) {
+					const bytes = await this.store.get(key);
+					if (!bytes) continue;
+					event = decodeEvent(bytes);
+					applyEvent(this.state, event);
+					this.pending.set(key, event);
+				}
+				toDelete.push(key);
+				if (!this.ledgerIds.has(event.id)) toAppend.push(event);
+			}
+			if (toDelete.length === 0) return { appended: 0, deleted: 0, backedUp: false };
+
+			let backedUp = false;
+			const day = this.clock().toISOString().slice(0, 10);
+			if (this.ledgerExists && this.lastBackupDay !== day) {
+				await this.store.copy(LEDGER_PATH, `backups/ledger-${day}.jsonl`);
+				this.lastBackupDay = day;
+				backedUp = true;
+			}
+
+			if (toAppend.length > 0) {
+				const bytes = encodeJsonl(toAppend);
+				if (this.ledgerExists) await this.store.append(LEDGER_PATH, bytes);
+				else await this.store.put(LEDGER_PATH, bytes, 'application/x-ndjson');
+				this.ledgerExists = true;
+				for (const e of toAppend) this.ledgerIds.add(e.id);
+			}
+
+			await this.store.delete(toDelete);
+			for (const key of toDelete) this.pending.delete(key);
+			return { appended: toAppend.length, deleted: toDelete.length, backedUp };
+		});
+	}
+
+	// ---- internals ---------------------------------------------------------
+
+	private run<T>(fn: () => Promise<T>): Promise<T> {
+		const next = this.queue.then(fn, fn);
+		this.queue = next.catch(() => undefined);
+		return next;
+	}
+
+	private requireExpense(id: string, expectedVersion: number): Expense {
+		const current = this.state.expenses.get(id);
+		if (!current) throw new NotFoundError(id);
+		if (current.version !== expectedVersion) throw new ConflictError(id, current.version);
+		return current;
+	}
+
+	private requireSettlement(id: string, expectedVersion: number): Settlement {
+		const current = this.state.settlements.get(id);
+		if (!current) throw new NotFoundError(id);
+		if (current.version !== expectedVersion) throw new ConflictError(id, current.version);
+		return current;
+	}
+
+	private async writeEvent(partial: NewEvent, actor: string, ts: string): Promise<void> {
+		if (this.readOnly) throw new ReadOnlyError();
+		const event = { ...partial, id: this.newId(), ts, actor } as LedgerEvent;
+		const key = eventKey(event.id);
+		await this.store.put(key, encodeEvent(event), 'application/json');
+		applyEvent(this.state, event);
+		this.pending.set(key, event);
+	}
+}

tests/unit/ledger/ledger.test.ts

@@ -0,0 +1,122 @@
+import { mkdtemp, rm } from 'node:fs/promises';
+import { tmpdir } from 'node:os';
+import { join } from 'node:path';
+import { afterEach, beforeEach, describe, expect, it } from 'vitest';
+import type { ExpenseBody, SettlementBody } from '../../../src/lib/domain/types';
+import { ConflictError, NotFoundError, ReadOnlyError } from '../../../src/lib/server/ledger/errors';
+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';
+
+const body: ExpenseBody = {
+	description: 'Pizza', amountCents: 2000, currency: 'USD', date: '2026-08-01', paidBy: 'u_a',
+	split: { mode: 'equal', participants: ['u_a', 'u_b'] }, shares: { u_a: 1000, u_b: 1000 }
+};
+const settlementBody: SettlementBody = { from: 'u_b', to: 'u_a', amountCents: 1000, date: '2026-08-02' };
+
+describe('Ledger', () => {
+	let dir: string;
+	let store: LocalFsStore;
+	let clockMs = Date.UTC(2026, 7, 29, 12, 0, 0);
+	const clock = () => new Date(clockMs);
+
+	beforeEach(async () => {
+		dir = await mkdtemp(join(tmpdir(), 'splitwise-ledger-'));
+		store = new LocalFsStore(dir);
+	});
+	afterEach(() => rm(dir, { recursive: true, force: true }));
+
+	it('loads empty from an empty store', async () => {
+		const ledger = await Ledger.load(store, { clock });
+		expect(ledger.listExpenses()).toEqual([]);
+		expect(ledger.stats()).toEqual({ pendingEvents: 0, ledgerEvents: 0, ledgerExists: false });
+	});
+
+	it('creates expenses, writes one event file each, and survives reload', async () => {
+		const ledger = await Ledger.load(store, { clock });
+		const created = await ledger.createExpense('u_a', body);
+		expect(created.version).toBe(1);
+		expect(created.createdAt).toBe(clock().toISOString());
+		expect(created.id).toMatch(/^e_[0-9A-Z]{26}$/);
+		const withId = await ledger.createExpense('u_a', body, 'e_CUSTOM');
+		expect(withId.id).toBe('e_CUSTOM');
+
+		const files = await store.list('events/');
+		expect(files).toHaveLength(2);
+		expect(JSON.parse(decodeText((await store.get(files[0].path))!)).entityId).toBe(created.id);
+
+		const reloaded = await Ledger.load(store, { clock });
+		expect(reloaded.getExpense(created.id)).toEqual(created);
+		expect(reloaded.stats().pendingEvents).toBe(2);
+	});
+
+	it('updates with version check and records history', async () => {
+		const ledger = await Ledger.load(store, { clock });
+		const e = await ledger.createExpense('u_a', body);
+		clockMs += 1000;
+		const updated = await ledger.updateExpense('u_b', e.id, { ...body, description: 'Pizza night' }, 1);
+		expect(updated.version).toBe(2);
+		expect(updated.updatedBy).toBe('u_b');
+		expect(updated.createdBy).toBe('u_a');
+		expect(updated.createdAt).toBe(e.createdAt);
+		await expect(ledger.updateExpense('u_a', e.id, body, 1)).rejects.toBeInstanceOf(ConflictError);
+		expect(ledger.history(e.id).map((ev) => ev.version)).toEqual([1, 2]);
+	});
+
+	it('deletes with version check; deleted items disappear but history remains', async () => {
+		const ledger = await Ledger.load(store, { clock });
+		const e = await ledger.createExpense('u_a', body);
+		await expect(ledger.deleteExpense('u_a', e.id, 5)).rejects.toBeInstanceOf(ConflictError);
+		await ledger.deleteExpense('u_a', e.id, 1);
+		expect(ledger.getExpense(e.id)).toBeUndefined();
+		await expect(ledger.deleteExpense('u_a', e.id, 2)).rejects.toBeInstanceOf(NotFoundError);
+		const reloaded = await Ledger.load(store, { clock });
+		expect(reloaded.getExpense(e.id)).toBeUndefined();
+		expect(reloaded.history(e.id)).toHaveLength(2);
+	});
+
+	it('handles settlements the same way', async () => {
+		const ledger = await Ledger.load(store, { clock });
+		const s = await ledger.createSettlement('u_b', settlementBody);
+		expect(s.id).toMatch(/^s_/);
+		const s2 = await ledger.updateSettlement('u_b', s.id, { ...settlementBody, amountCents: 500 }, 1);
+		expect(s2.amountCents).toBe(500);
+		await ledger.deleteSettlement('u_b', s.id, 2);
+		expect(ledger.listSettlements()).toEqual([]);
+	});
+
+	it('refuses writes in read-only mode and leaves state untouched on store failure', async () => {
+		const ledger = await Ledger.load(store, { clock });
+		ledger.readOnly = true;
+		await expect(ledger.createExpense('u_a', body)).rejects.toBeInstanceOf(ReadOnlyError);
+		ledger.readOnly = false;
+
+		const failing = new LocalFsStore(dir);
+		failing.put = async () => {
+			throw new Error('network down');
+		};
+		const l2 = await Ledger.load(failing, { clock });
+		await expect(l2.createExpense('u_a', body)).rejects.toThrow('network down');
+		expect(l2.listExpenses()).toEqual([]);
+		expect(l2.stats().pendingEvents).toBe(0);
+	});
+
+	it('serialises concurrent writes', async () => {
+		const ledger = await Ledger.load(store, { clock });
+		const results = await Promise.all(Array.from({ length: 10 }, (_, i) => ledger.createExpense('u_a', { ...body, description: `e${i}` })));
+		expect(new Set(results.map((r) => r.id)).size).toBe(10);
+		expect((await store.list('events/')).length).toBe(10);
+	});
+
+	it('loads the ledger file plus pending events, deduplicating', async () => {
+		const ledger = await Ledger.load(store, { clock });
+		const e = await ledger.createExpense('u_a', body);
+		const [file] = await store.list('events/');
+		const eventBytes = (await store.get(file.path))!;
+		// Simulate a compaction that appended but crashed before deleting the event file.
+		await store.put('ledger.jsonl', encodeText(decodeText(eventBytes) + '\n'));
+		const reloaded = await Ledger.load(store, { clock });
+		expect(reloaded.getExpense(e.id)).toEqual(e);
+		expect(reloaded.stats()).toEqual({ pendingEvents: 1, ledgerEvents: 1, ledgerExists: true });
+	});
+});