Feat/get aggregates

package.json

@@ -5,7 +5,7 @@
   "types": "index.d.ts",
   "main": "index.js",
   "scripts": {
-    "test": "mocha \"src/test/**/*.spec.js\" --exit -t 10000",
+    "test": "npx ts-mocha \"src/test/**/*.spec.ts\" --exit -t 10000",
     "clean": "rimraf \"*.js\" \"*.js.map\" \"*.d.ts\" \"src/**/*.js*\" \"src/**/*.d.ts\" \"provider/**/*.js*\" \"provider/**/*.d.ts\"",
     "prebuild": "yarn clean",
     "build": "tsc",
@@ -29,23 +29,26 @@
   "bugs": {
     "url": "https://github.com/seikho/evtstore/issues"
   },
-  "dependencies": {},
   "devDependencies": {
     "@types/chai": "^4.2.3",
     "@types/dotenv": "^6.1.1",
+    "@types/expect": "^1.20.4",
     "@types/mocha": "^5.2.7",
+    "@types/node": "^22.15.30",
     "@types/pg": "^8.6.1",
     "@types/sqlite3": "^3.1.5",
     "chai": "^4.2.0",
     "dotenv": "^8.1.0",
     "knex": "^2.4.0",
-    "mocha": "^10.2.0",
+    "mocha": "^10.8.2",
     "mongodb": "^4.11.0",
     "neo4j-driver": "^4.1.2",
     "pg": "^8.2.1",
     "postgres": "^3.2.4",
     "rimraf": "^3.0.0",
     "sqlite3": "^5.1.4",
-    "typescript": "^4.9.4"
+    "ts-mocha": "^11.1.0",
+    "ts-node": "^10.9.2",
+    "typescript": "^4.9.5"
   }
 }

provider/knex.ts

@@ -87,6 +87,21 @@ export function createProvider<E extends Event>(opts: Options): Provider<E> {
 
       return events.map(mapToEvent)
     },
+    getBatchEventsFor: async (stream, aggregateIds) => {
+      const query = opts
+        .events()
+        .select()
+        .whereIn('aggregate_id', toArray(aggregateIds))
+        .andWhere('stream', stream)
+        .orderBy([
+          { column: 'timestamp', order: 'asc' },
+          { column: 'version', order: 'asc' },
+        ])
+
+      const events = await query
+
+      return events.map(mapToEvent)
+    },
     createEvents: createEventsMapper<E>(0),
     append: async (_stream, _aggregateId, _version, newEvents) => {
       try {

provider/memory.ts

@@ -41,6 +41,10 @@ export function createProvider<E extends Event>(
     return events.filter(filter)
   }
 
+  const getBatchEventsFor = async (stream: string, aggregateIds: string[]) => {
+    return events.filter((ev) => ev.stream === stream && aggregateIds.includes(ev.aggregateId))
+  }
+
   const createEvents = createEventsMapper<E>(0)
 
   const append = async (
@@ -71,6 +75,7 @@ export function createProvider<E extends Event>(
     getEventsFor,
     getEventsFrom,
     getLastEventFor,
+    getBatchEventsFor,
     createEvents,
     append,
   }

provider/mongo.ts

@@ -76,6 +76,17 @@ export function createProvider<E extends Event>(opts: Options<E>): Provider<E> {
         return query.toArray()
       }),
 
+    getBatchEventsFor: async (stream, ids) => {
+      const query = {
+        stream,
+        aggregateId: { $in: ids },
+      } as unknown as Filter<StoreEvent<E>>
+
+      const results = await events.then((coll) => coll.find(query).sort({ position: 1 }).toArray())
+
+      return results
+    },
+
     createEvents,
 
     append: async (_stream, _aggId, _version, newEvents) => {

provider/neo4j-v3.ts

@@ -132,6 +132,30 @@ export function createProvider<E extends Event>(opts: Options): Provider<E> {
 
       return parsed
     },
+    getBatchEventsFor: async (stream, aggregateIds) => {
+      const params: any = { stream, aggregateIds }
+
+      const query = `
+        MATCH (ev: ${opts.events})
+        WHERE ev.stream = $stream
+        AND ev.aggregateId IN $aggregateIds
+        RETURN ev
+        ORDER BY ev.timestamp ASC, ev.version ASC
+      `
+
+      const events = await run<any>(query, params)
+
+      const parsed = events.map((ev) => ({
+        stream: ev.stream,
+        position: toInternalPosition(ev.position),
+        version: toVersion(ev.version),
+        timestamp: new Date(ev.timestamp),
+        aggregateId: ev.aggregateId,
+        event: JSON.parse(ev.event),
+      }))
+
+      return parsed
+    },
     createEvents: createEventsMapper<E>(0),
     append: async (stream, id, _version, newEvents) => {
       const client = await opts.client

provider/neo4j.ts

@@ -128,6 +128,30 @@ export function createProvider<E extends Event>(opts: Options): Provider<E> {
 
       return parsed
     },
+    getBatchEventsFor: async (stream, aggregateIds) => {
+      const params: any = { stream, aggregateIds }
+
+      const query = `
+        MATCH (ev: ${opts.events})
+        WHERE ev.stream = $stream
+        AND ev.aggregateId IN $aggregateIds
+        RETURN ev
+        ORDER BY ev.timestamp ASC, ev.version ASC
+      `
+
+      const events = await run<any>(query, params)
+
+      const parsed = events.map((ev) => ({
+        stream: ev.stream,
+        position: toInternalPosition(ev.position),
+        version: toVersion(ev.version),
+        timestamp: new Date(ev.timestamp),
+        aggregateId: ev.aggregateId,
+        event: JSON.parse(ev.event),
+      }))
+
+      return parsed
+    },
     createEvents: createEventsMapper<E>(0),
     append: async (stream, id, _version, newEvents) => {
       const client = await opts.client

provider/pg.ts

@@ -101,6 +101,16 @@ export function createProvider<E extends Event>(opts: Options): Provider<E> {
       const result = await opts.client.query(q, values)
       return result.rows.map(mapToEvent)
     },
+    getBatchEventsFor: async (stream, aggregateIds) => {
+      const placeholders = aggregateIds.map((_, i) => `$${i + 2}`).join(', ')
+      let query = `select * from "${opts.events}" where stream = $1 and aggregate_id in (${placeholders})`
+      const values = [stream, ...aggregateIds]
+
+      query += ` order by timestamp, version asc`
+
+      const result = await opts.client.query(query, values)
+      return result.rows.map(mapToEvent)
+    },
     createEvents: createEventsMapper<E>(0),
     append: async (_stream, _aggregateId, _version, newEvents) => {
       const trx = await opts.client.connect()

provider/postgres.ts

@@ -60,7 +60,7 @@ export function createProvider<E extends Event>(opts: Options): Provider<E> {
     getLastEventFor: async (stream, aggregateId) => {
       const streams = Array.isArray(stream) ? stream : [stream]
       const agg = aggregateId ? sql`AND aggregate_id = ${aggregateId}` : sql``
-      const result = await sql`select * from ${sql(evts)} where stream in (${sql(
+      const result = await sql`select * from ${sql(evts)} WHERE stream = ANY(${sql.array(
         streams
       )}) ${agg} order by position desc limit 1`
 
@@ -71,9 +71,18 @@ export function createProvider<E extends Event>(opts: Options): Provider<E> {
       const limit = lim ?? opts.limit
       const limitClause = limit ? sql`LIMIT ${limit}` : sql``
 
-      const result = await sql`SELECT * FROM ${sql(evts)} WHERE stream IN ${sql(
+      const result = await sql`SELECT * FROM ${sql(evts)} WHERE stream = ANY(${sql.array(
         streams
-      )} AND position > ${position} ORDER BY position ASC ${limitClause}`
+      )}) AND position > ${position} ORDER BY position ASC ${limitClause}`
+
+      return result.map(mapToEvent)
+    },
+    getBatchEventsFor: async (stream, aggregateIds) => {
+      const result = await sql`SELECT * FROM ${sql(evts)}
+        WHERE stream = ${stream}
+        AND aggregate_id IN ${sql(aggregateIds)}
+        ORDER BY timestamp, version asc
+      `
 
       return result.map(mapToEvent)
     },

src/create-aggregate.ts

@@ -74,7 +74,42 @@ export function createProvidedAggregate<E extends Event, A extends Aggregate>(
     }
   }
 
-  return { stream: opts.stream, getAggregate, toNextAggregate, provider: opts.provider }
+  async function getAggregates(ids: string[]) {
+    const events = await (await opts.provider).getBatchEventsFor(opts.stream, ids)
+
+    const aggMap = new Map<string, A & BaseAggregate>()
+
+    for (const id of ids) {
+      aggMap.set(id, { ...opts.aggregate(), aggregateId: id, version: 0 })
+    }
+
+    for (const ev of events) {
+      const agg = aggMap.get(ev.aggregateId)
+
+      if (agg) {
+        aggMap.set(ev.aggregateId, toNextAggregate(agg, ev))
+      }
+    }
+
+    const aggregates: (A & BaseAggregate)[] = []
+
+    for (const id of ids) {
+      const agg = aggMap.get(id)
+      if (agg) {
+        aggregates.push(agg)
+      }
+    }
+
+    return aggregates
+  }
+
+  return {
+    stream: opts.stream,
+    getAggregate,
+    getAggregates,
+    toNextAggregate,
+    provider: opts.provider,
+  }
 }
 
 export function createPersistedAggregate<E extends Event, A extends Aggregate>(
@@ -147,9 +182,39 @@ export function createPersistedAggregate<E extends Event, A extends Aggregate>(
     }
   }
 
+  async function getAggregates(ids: string[]) {
+    const events = await (await opts.provider).getBatchEventsFor(opts.stream, ids)
+
+    const aggMap = new Map<string, A & BaseAggregate>()
+
+    for (const id of ids) {
+      aggMap.set(id, { ...opts.aggregate(), aggregateId: id, version: 0 })
+    }
+
+    for (const ev of events) {
+      const agg = aggMap.get(ev.aggregateId)
+
+      if (agg) {
+        aggMap.set(ev.aggregateId, toNextAggregate(agg, ev))
+      }
+    }
+
+    const aggregates: (A & BaseAggregate)[] = []
+
+    for (const id of ids) {
+      const agg = aggMap.get(id)
+      if (agg) {
+        aggregates.push(agg)
+      }
+    }
+
+    return aggregates
+  }
+
   return {
     stream: opts.stream,
     getAggregate: getPersistedAggregate,
+    getAggregates,
     toNextAggregate,
     provider: opts.provider,
     version: opts.version,

src/domain.ts

@@ -44,7 +44,7 @@ function wrapCmd<E extends Event, A extends Aggregate, C extends Command>(
     throw new Error(`Invalid command body: Command handler function cannot be named "aggregate"`)
   }
 
-  const { getAggregate, toNextAggregate } = createProvidedAggregate<E, A>(opts)
+  const { getAggregate, toNextAggregate, getAggregates } = createProvidedAggregate<E, A>(opts)
 
   async function getExecAggregate(id: string) {
     const aggregate = await getAggregate(id)
@@ -65,6 +65,31 @@ function wrapCmd<E extends Event, A extends Aggregate, C extends Command>(
     return { ...body, aggregate }
   }
 
+  async function getExecAggregates(ids: string[]) {
+    const aggregates = await getAggregates(ids)
+
+    const bodies: Array<ExecutableAggregate<C, A> & { aggregate: A & BaseAggregate }> = []
+
+    for (const aggregate of aggregates) {
+      const body: ExecutableAggregate<C, A> = {} as any
+
+      for (const command of commands) {
+        body[command] = async (cmdBody) => {
+          const cmdResult = await handler[command](
+            { ...cmdBody, aggregateId: aggregate.aggregateId, type: command },
+            aggregate
+          )
+          const nextAggregate = await handleCommandResult(cmdResult, aggregate)
+
+          return { ...body, aggregate: nextAggregate }
+        }
+      }
+      bodies.push({ ...body, aggregate })
+    }
+
+    return bodies
+  }
+
   // Prepare the command handlers that accept an aggregateId and a command body
   for (const type of commands) {
     wrapped[type] = async (id, body) => {
@@ -94,5 +119,5 @@ function wrapCmd<E extends Event, A extends Aggregate, C extends Command>(
     return nextAggregate
   }
 
-  return { command: wrapped, getAggregate: getExecAggregate }
+  return { command: wrapped, getAggregate: getExecAggregate, getAggregates: getExecAggregates }
 }

src/test/provider.spec.ts

@@ -37,6 +37,33 @@ describe('provider tests', () => {
         match({ one: 84, version: 2 }, actual)
       })
 
+      it('will get batch aggregates', async () => {
+        await domain.command.doOne('id1', { one: 1 })
+        await domain.command.doTwo('id1', { two: 'two' })
+        await domain.command.doThree('id1', { three: [3] })
+
+        await domain.command.doOne('id2', { one: 100 })
+        await domain.command.doTwo('id2', { two: 'number two' })
+
+        await domain.command.doOne('id3', { one: 111 })
+        await domain.command.doThree('id3', { three: [333] })
+
+        const aggs = await domain.getAggregates(['id2', 'id1', 'id3'])
+        expect(aggs.length).to.equal(3)
+
+        // Intentionally out of order.
+        const [aggWithId2, aggWithId1, aggWithId3] = aggs
+
+        match({ aggregateId: 'id1', one: 1, two: 'two', version: 3 }, aggWithId1.aggregate)
+        expect(aggWithId1.aggregate.three[0]).to.equal(3)
+
+        match({ aggregateId: 'id2', one: 100, two: 'number two', version: 2 }, aggWithId2.aggregate)
+        expect(aggWithId2.aggregate.three[0]).to.be.undefined
+
+        match({ aggregateId: 'id3', one: 111, two: '', version: 2 }, aggWithId3.aggregate)
+        expect(aggWithId3.aggregate.three[0]).to.equal(333)
+      })
+
       it('will correctly update model using event handler', async () => {
         await domain.populator.runOnce()
         const actual = domain.models.get('one')
@@ -120,12 +147,12 @@ describe('provider tests', () => {
           ++count
         })
         await pop.runOnce()
-        expect(count).to.equal(6)
+        expect(count).to.equal(9)
         await pop.runOnce()
-        expect(count).to.equal(6)
+        expect(count).to.equal(9)
         await domain.command.doOne('in-memory', { one: 1 })
         await pop.runOnce()
-        expect(count).to.equal(7)
+        expect(count).to.equal(10)
       })
 
       it('will correctly handle multiple streams in a single handler', async () => {