Implement event store caching (#27)

Author: lazamar

package-lock.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"

package.json

@@ -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 = <E, T>(
   onError: (e: Error) => E,
   f: (s: EventStore) => Future<E, T>,
 ) => Future<E, T>;
 
+type LoadedAggregate<T extends Aggregate<T>> = {
+  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<Id<Aggregate<unknown>>, LoadedAggregate<any>>;
+
+  // 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<T extends Aggregate<T>>(
     cls: Constructor<T>,
     aggregateId: Id<T>,
-  ): Promise<{ aggregate: T; lastEvent: EventInfo }> {
+  ): Promise<T> {
+    return (await this._find(cls, aggregateId)).aggregate;
+  }
+
+  async try_find<T extends Aggregate<T>>(
+    cls: Constructor<T>,
+    aggregateId: Id<T>,
+  ): Promise<T | null> {
+    const found = await this._try_find(cls, aggregateId);
+    return found ? found.aggregate : null;
+  }
+
+  private async _find<T extends Aggregate<T>>(
+    cls: Constructor<T>,
+    aggregateId: Id<T>,
+  ): Promise<LoadedAggregate<T>> {
+    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<T extends Aggregate<T>>(
+    cls: Constructor<T>,
+    aggregateId: Id<T>,
+  ): Promise<Nullable<LoadedAggregate<T>>> {
+    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<T extends Aggregate<T>>(args: {
@@ -50,13 +104,14 @@ class PostgresEventStore implements EventStore {
     event_id?: Id<Event<T>>;
     correlation_id?: Id<Event<T>>;
     causation_id?: Id<Event<T>>;
-  }): Promise<void> {
+  }): Promise<{ event: Event<T>; 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<Event<T>>({ info, event });
+    this.cache_save({ aggregate, lastEvent: info });
+    return { event, info };
   }
 
   async doesEventAlreadyExist(eventId: Id<Event<any>>): Promise<boolean> {
@@ -157,6 +215,16 @@ class PostgresEventStore implements EventStore {
       throw new Error(`Failed to save event: ${edata.info.event_id}: ${error}`);
     }
   }
+
+  private cache_save<T extends Aggregate<T>>(loaded: LoadedAggregate<T>): void {
+    this.cache.set(loaded.aggregate.aggregateId, loaded);
+  }
+
+  private cache_load<T extends Aggregate<T>>(
+    id: Id<T>,
+  ): Nullable<LoadedAggregate<T>> {
+    return this.cache.get(id).asNullable();
+  }
 }
 
 // Prepare the database to be used as an event store.

src/app/eventStore.ts

@@ -19,10 +19,7 @@ const handler: ReactionHandler<Events> = ({
   store,
 }): Future<AmbarResponse, void> =>
   Future.attemptP<void>(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;

src/domain/cookingClub/membership/reaction/evaluateApplication.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 = <T>(): TreeMap<string, T> =>
+  TreeMap.new((x: string, y: string) => (x > y ? 1 : x < y ? -1 : 0));
+
+interface Comparable<T> {
+  compare(other: T): number;
+}
+
+// This is just a wrapper around BTree which requires
+// the comparison function.
+class TreeMap<K, V> {
+  // @ts-expect-error: unused _. Prevent instantiation by casting.
+  private readonly _: null = null;
+  tree: BTree<K, V>;
+  compare: (l: K, r: K) => number;
+
+  static new<K, V>(compare: (l: K, r: K) => number): TreeMap<K, V> {
+    return new TreeMap(compare);
+  }
+
+  static new_<K extends Comparable<K>, V>(): TreeMap<K, V> {
+    const compare = (x: K, y: K) => x.compare(y);
+    return new TreeMap(compare);
+  }
+
+  static from<K, V>(
+    compare: (l: K, r: K) => number,
+    xs: Array<[K, V]>,
+  ): TreeMap<K, V> {
+    return TreeMap.new<K, V>(compare).setEntries(xs[Symbol.iterator]());
+  }
+
+  static from_<K extends Comparable<K>, V>(xs: Array<[K, V]>): TreeMap<K, V> {
+    return TreeMap.new_<K, V>().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<V> {
+    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<K, V> {
+    this.tree.delete(k);
+    return this;
+  }
+
+  keys(): IterableIterator<K> {
+    return this.tree.keys();
+  }
+
+  values(): IterableIterator<V> {
+    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<K, V>) {
+    for (const [k, v] of other.entries()) {
+      this.set(k, v);
+    }
+    return this;
+  }
+
+  unionWith(other: TreeMap<K, V>, 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<W, X>(
+    other: TreeMap<K, W>,
+    f: (left: V, right: W) => X,
+  ): TreeMap<K, X> {
+    const result = new TreeMap<K, X>(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<K, unknown>): TreeMap<K, V> {
+    const diff = new TreeMap<K, V>(this.compare);
+    for (const [k, v] of this.entries()) {
+      if (!other.has(k)) {
+        diff.set(k, v);
+      }
+    }
+    return diff;
+  }
+
+  mapWithKeys<W>(f: (k: K, v: V) => W): TreeMap<K, W> {
+    const n = new TreeMap<K, W>(this.compare);
+    for (const [k, v] of this.entries()) {
+      n.set(k, f(k, v));
+    }
+    return n;
+  }
+
+  map<W>(f: (v: V) => W): TreeMap<K, W> {
+    return this.mapWithKeys((_, v) => f(v));
+  }
+
+  size(): number {
+    return this.tree.size;
+  }
+}