From b6420105e73295ae8b2b0103aa6c58eb330954fd Mon Sep 17 00:00:00 2001 From: Marcelo Lazaroni Date: Wed, 15 Oct 2025 12:25:27 +0100 Subject: [PATCH 1/3] Introduce TreeMap --- src/lib/TreeMap.ts | 154 +++++++++++++++++++++++++++++++++++++++++++++ 1 file changed, 154 insertions(+) create mode 100644 src/lib/TreeMap.ts diff --git a/src/lib/TreeMap.ts b/src/lib/TreeMap.ts new file mode 100644 index 0000000..d767446 --- /dev/null +++ b/src/lib/TreeMap.ts @@ -0,0 +1,154 @@ +/* + A Map type that requires a comparison function +*/ +export { TreeMap, stringMap }; + +import BTree from 'sorted-btree'; + +import { Maybe, Just, Nothing } from '@/lib/Maybe'; + +const stringMap = (): TreeMap => + TreeMap.new((x: string, y: string) => (x > y ? 1 : x < y ? -1 : 0)); + +interface Comparable { + compare(other: T): number; +} + +// This is just a wrapper around BTree which requires +// the comparison function. +class TreeMap { + // @ts-expect-error: unused _. Prevent instantiation by casting. + private readonly _: null = null; + tree: BTree; + compare: (l: K, r: K) => number; + + static new(compare: (l: K, r: K) => number): TreeMap { + return new TreeMap(compare); + } + + static new_, V>(): TreeMap { + const compare = (x: K, y: K) => x.compare(y); + return new TreeMap(compare); + } + + static from( + compare: (l: K, r: K) => number, + xs: Array<[K, V]>, + ): TreeMap { + return TreeMap.new(compare).setEntries(xs[Symbol.iterator]()); + } + + static from_, V>(xs: Array<[K, V]>): TreeMap { + return TreeMap.new_().setEntries(xs[Symbol.iterator]()); + } + + private constructor(compare: (l: K, r: K) => number) { + this.compare = compare; + this.tree = new BTree([], compare); + } + + set(k: K, v: V) { + this.tree.set(k, v); + return this; + } + + setWith(k: K, v: V, f: (old: V, _new: V) => V) { + const found = this.tree.get(k); + if (found !== undefined) { + this.tree.set(k, f(found, v)); + } else { + this.tree.set(k, v); + } + return this; + } + + get(k: K): Maybe { + const found = this.tree.get(k); + return found !== undefined ? Just(found) : Nothing(); + } + + has(k: K): boolean { + return this.tree.has(k); + } + + remove(k: K): TreeMap { + this.tree.delete(k); + return this; + } + + keys(): IterableIterator { + return this.tree.keys(); + } + + values(): IterableIterator { + return this.tree.values(); + } + + entries(): IterableIterator<[K, V]> { + return this.tree.entries(); + } + + setEntries(it: IterableIterator<[K, V]>) { + for (const [k, v] of it) { + this.set(k, v); + } + return this; + } + + union(other: TreeMap) { + for (const [k, v] of other.entries()) { + this.set(k, v); + } + return this; + } + + unionWith(other: TreeMap, f: (old: V, new_: V) => V) { + for (const [k, v] of other.entries()) { + this.setWith(k, v, f); + } + return this; + } + + // Create a new TreeMap from keys common to two other maps. + intersectionWith( + other: TreeMap, + f: (left: V, right: W) => X, + ): TreeMap { + const result = new TreeMap(this.compare); + for (const [k, v] of this.entries()) { + const found = other.get(k); + if (found instanceof Just) { + result.set(k, f(v, found.value)); + } + } + return result; + } + + // Difference in the set of keys + // A.difference(B) equals A minus all keys present in B. + difference(other: TreeMap): TreeMap { + const diff = new TreeMap(this.compare); + for (const [k, v] of this.entries()) { + if (!other.has(k)) { + diff.set(k, v); + } + } + return diff; + } + + mapWithKeys(f: (k: K, v: V) => W): TreeMap { + const n = new TreeMap(this.compare); + for (const [k, v] of this.entries()) { + n.set(k, f(k, v)); + } + return n; + } + + map(f: (v: V) => W): TreeMap { + return this.mapWithKeys((_, v) => f(v)); + } + + size(): number { + return this.tree.size; + } +} From dc0dc309af649bf097c1638bc4ce2f25aa7b9ac2 Mon Sep 17 00:00:00 2001 From: Marcelo Lazaroni Date: Wed, 15 Oct 2025 12:25:55 +0100 Subject: [PATCH 2/3] Introduce TreeSet --- src/lib/TreeSet.ts | 78 ++++++++++++++++++++++++++++++++++++++++++++++ 1 file changed, 78 insertions(+) create mode 100644 src/lib/TreeSet.ts diff --git a/src/lib/TreeSet.ts b/src/lib/TreeSet.ts new file mode 100644 index 0000000..bb93a01 --- /dev/null +++ b/src/lib/TreeSet.ts @@ -0,0 +1,78 @@ +/* + A Set type that requires a comparison function +*/ +export { TreeSet }; + +import BTree from 'sorted-btree'; + +interface Comparable { + compare(other: T): number; +} + +// This is just a wrapper around BTree which requires +// the comparison function. +class TreeSet { + // @ts-expect-error: unused _. Prevent instantiation by casting. + private readonly _: null = null; + tree: BTree; + compare: (l: K, r: K) => number; + + static new(compare: (l: K, r: K) => number): TreeSet { + return new TreeSet(compare); + } + + static new_>(): TreeSet { + const compare = (x: K, y: K) => x.compare(y); + return new TreeSet(compare); + } + + // Return a clone of the tree. + static from(tree: TreeSet): TreeSet { + const created = TreeSet.new(tree.compare); + created.tree = tree.tree.clone(); + return created; + } + + static from_>(xs: Array): TreeSet { + return TreeSet.new_().insertValues(xs); + } + + private constructor(compare: (l: K, r: K) => number) { + this.compare = compare; + this.tree = new BTree([], compare); + } + + insert(k: K): TreeSet { + this.tree.set(k, null); + return this; + } + + has(k: K): boolean { + return this.tree.has(k); + } + + remove(k: K): TreeSet { + this.tree.delete(k); + return this; + } + + values(): Array { + return Array.from(this.tree.keys()); + } + + insertValues(it: Array) { + for (const k of it) { + this.insert(k); + } + return this; + } + + union(other: TreeSet) { + this.insertValues(other.values()); + return this; + } + + size(): number { + return this.tree.size; + } +} From 9a140867b9f3afd23872acb5a04f9d7430d5d7ae Mon Sep 17 00:00:00 2001 From: Marcelo Lazaroni Date: Wed, 15 Oct 2025 12:58:23 +0100 Subject: [PATCH 3/3] Implement event store caching --- package-lock.json | 7 ++ package.json | 1 + src/app/eventStore.ts | 90 ++++++++++++++++--- .../reaction/evaluateApplication.ts | 5 +- src/lib/eventSourcing/event.ts | 4 + src/lib/eventSourcing/eventStore.ts | 9 +- 6 files changed, 99 insertions(+), 17 deletions(-) diff --git a/package-lock.json b/package-lock.json index 11c8d06..f77d7d8 100644 --- a/package-lock.json +++ b/package-lock.json @@ -18,6 +18,7 @@ "nodemailer": "^7.0.5", "pg": "^8.8.0", "reflect-metadata": "^0.2.2", + "sorted-btree": "^1.8.1", "tsyringe": "^4.8.0", "winston": "^3.8.2", "zod": "^3.24.1" @@ -4306,6 +4307,12 @@ "npm": ">= 3.0.0" } }, + "node_modules/sorted-btree": { + "version": "1.8.1", + "resolved": "https://registry.npmjs.org/sorted-btree/-/sorted-btree-1.8.1.tgz", + "integrity": "sha512-395+XIP+wqNn3USkFSrNz7G3Ss/MXlZEqesxvzCRFwL14h6e8LukDHdLBePn5pwbm5OQ9vGu8mDyz2lLDIqamQ==", + "license": "MIT" + }, "node_modules/sparse-bitfield": { "version": "3.0.3", "resolved": "https://registry.npmjs.org/sparse-bitfield/-/sparse-bitfield-3.0.3.tgz", diff --git a/package.json b/package.json index 8ef9749..6d87bfd 100644 --- a/package.json +++ b/package.json @@ -21,6 +21,7 @@ "nodemailer": "^7.0.5", "pg": "^8.8.0", "reflect-metadata": "^0.2.2", + "sorted-btree": "^1.8.1", "tsyringe": "^4.8.0", "winston": "^3.8.2", "zod": "^3.24.1" diff --git a/src/app/eventStore.ts b/src/app/eventStore.ts index 0c56f2a..39601c6 100644 --- a/src/app/eventStore.ts +++ b/src/app/eventStore.ts @@ -19,29 +19,83 @@ import { PostgresTransaction } from '@/lib/postgres'; import { log } from '@/common/util/Logger'; import { POSIX } from '@/lib/time'; import { Future } from '@/lib/Future'; +import { TreeMap } from '@/lib/TreeMap'; +import { Nullable } from '@/lib/Maybe'; type WithEventStore = ( onError: (e: Error) => E, f: (s: EventStore) => Future, ) => Future; +type LoadedAggregate> = { + aggregate: T; + lastEvent: EventInfo; +}; + class PostgresEventStore implements EventStore { + // This cache allows us to efficiently call `find` and `try_find` multiple + // times within a transaction. This makes reactions and commands simpler as + // there is no need to manually apply to the aggregate the transformations + // performed by newly emitted events in those functions. Instead we can just + // call `find` again and load the latest version of the aggregate for free. + private cache: TreeMap>, LoadedAggregate>; + + // An instance of this class never lives loger than the transaction + // it is associated with. constructor( private transaction: PostgresTransaction, private readonly schemas: Schemas, private readonly eventStoreTable: string, - ) {} + ) { + this.cache = TreeMap.new_(); + } async find>( cls: Constructor, aggregateId: Id, - ): Promise<{ aggregate: T; lastEvent: EventInfo }> { + ): Promise { + return (await this._find(cls, aggregateId)).aggregate; + } + + async try_find>( + cls: Constructor, + aggregateId: Id, + ): Promise { + const found = await this._try_find(cls, aggregateId); + return found ? found.aggregate : null; + } + + private async _find>( + cls: Constructor, + aggregateId: Id, + ): Promise> { + const found = await this._try_find(cls, aggregateId); + + if (found == null) { + throw new Error(`Unknown aggregate ID ${aggregateId.value}`); + } + + return found; + } + + private async _try_find>( + cls: Constructor, + aggregateId: Id, + ): Promise>> { + const found = this.cache_load(aggregateId); + if (found !== null) { + return found; + } + const events = await this.findAll(aggregateId); - const { lastEvent, aggregate } = this.schemas - .hydrate(cls, events) - .unwrap((e) => e); - return { aggregate, lastEvent }; + if (events.length === 0) { + return null; + } + + const loaded = this.schemas.hydrate(cls, events).unwrap((e) => e); + this.cache_save(loaded); + return loaded; } async emit>(args: { @@ -50,13 +104,14 @@ class PostgresEventStore implements EventStore { event_id?: Id>; correlation_id?: Id>; causation_id?: Id>; - }): Promise { + }): Promise<{ event: Event; info: EventInfo }> { const event = args.event; const event_id = args.event_id || Id.random(); let info: EventInfo; + let aggregate: T; switch (true) { case event instanceof CreationEvent: { - const aggregate: T = event.createAggregate(); + aggregate = event.createAggregate(); info = { event_id, aggregate_id: aggregate.aggregateId, @@ -68,16 +123,17 @@ class PostgresEventStore implements EventStore { break; } case event instanceof TransformationEvent: { - const { aggregate, lastEvent } = await this.find( + const found = await this._find( args.aggregate, event.values.aggregateId, ); + aggregate = found.aggregate; info = { event_id, aggregate_id: aggregate.aggregateId, aggregate_version: aggregate.aggregateVersion + 1, - correlation_id: lastEvent.correlation_id, - causation_id: lastEvent.causation_id, + correlation_id: found.lastEvent.correlation_id, + causation_id: found.lastEvent.causation_id, recorded_on: POSIX.now(), }; break; @@ -87,6 +143,8 @@ class PostgresEventStore implements EventStore { } await this.insert>({ info, event }); + this.cache_save({ aggregate, lastEvent: info }); + return { event, info }; } async doesEventAlreadyExist(eventId: Id>): Promise { @@ -157,6 +215,16 @@ class PostgresEventStore implements EventStore { throw new Error(`Failed to save event: ${edata.info.event_id}: ${error}`); } } + + private cache_save>(loaded: LoadedAggregate): void { + this.cache.set(loaded.aggregate.aggregateId, loaded); + } + + private cache_load>( + id: Id, + ): Nullable> { + return this.cache.get(id).asNullable(); + } } // Prepare the database to be used as an event store. diff --git a/src/domain/cookingClub/membership/reaction/evaluateApplication.ts b/src/domain/cookingClub/membership/reaction/evaluateApplication.ts index 9431e09..afa559d 100644 --- a/src/domain/cookingClub/membership/reaction/evaluateApplication.ts +++ b/src/domain/cookingClub/membership/reaction/evaluateApplication.ts @@ -19,10 +19,7 @@ const handler: ReactionHandler = ({ store, }): Future => Future.attemptP(async () => { - const { aggregate: membership } = await store.find( - Membership, - event.values.aggregateId, - ); + const membership = await store.find(Membership, event.values.aggregateId); if (membership.status !== 'Requested') { return; diff --git a/src/lib/eventSourcing/event.ts b/src/lib/eventSourcing/event.ts index 1776b76..a09a0c4 100644 --- a/src/lib/eventSourcing/event.ts +++ b/src/lib/eventSourcing/event.ts @@ -51,6 +51,10 @@ class Id { const cleanId = base64Encoded.replace(/[^A-Za-z0-9]/g, ''); return cleanId.substring(0, ID_LENGTH); } + + compare(other: Id) { + return this.value > other.value ? 1 : this.value === other.value ? 0 : -1; + } } // Class which all events derive from. Used for type constraints. diff --git a/src/lib/eventSourcing/eventStore.ts b/src/lib/eventSourcing/eventStore.ts index e242184..42ded20 100644 --- a/src/lib/eventSourcing/eventStore.ts +++ b/src/lib/eventSourcing/eventStore.ts @@ -55,7 +55,12 @@ interface EventStore { find>( cls: Constructor, aggregateId: Id, - ): Promise<{ aggregate: T; lastEvent: EventInfo }>; + ): Promise; + + try_find>( + cls: Constructor, + aggregateId: Id, + ): Promise; emit>(args: { aggregate: Constructor; @@ -63,7 +68,7 @@ interface EventStore { event_id?: Id>; correlation_id?: Id>; causation_id?: Id>; - }): Promise; + }): Promise<{ event: Event; info: EventInfo }>; doesEventAlreadyExist(eventId: Id>): Promise; }