Type-safe Event Sourcing and CQRS with Node.JS and TypeScript
Note: createDomain will be migrating to createDomainV2 in version 11.x
The createDomainV2 API solves circular reference issues when importing aggregates.
The original createDomain will be available as createDomainV1 from 11.x onwards.
I reguarly use event sourcing and wanted to lower the barrier for entry and increase productivity for colleagues.
The design goals were:
- Provide as much type safety and inference as possible
- Make creating domains quick and intuitive
- Be easy to test
- Allow developers to focus on application/business problems instead of Event Sourcing and CQRS problems
To obtain these goals the design is highly opinionated, but still flexible.
See Providers for more details and examples
- Postgres using Postgres.js
- Postgres using node-postgres
- SQLite, MySQL, Postgres using Knex
- In-memory
- MongoDB
- Neo4j v3.5
- Neo4j v4
See the documentation regarding information about aggregate persistence. This refers to persisting a copy of the aggregate on events for performant retrieval.
EvtStore is type-driven to take advantage of type safety and auto completion. We front-load the creation of our Event, Aggregate, and Command types to avoid having to repeatedly import and pass them as generic argument. EvtStore makes use for TypeScript's mapped types and conditional types to achieve this.
typeUserEvt=|{type: 'created',name: string}|{type: 'disabled'}|{type: 'enabled'}typeUserAgg={name: string,enabled: boolean}typeUserCmd=|{type: 'create': name: string }|{type: 'enable'}|{type: 'disable'}typePostEvt=|{type: 'postCreated',userId: string,content: string}|{type: 'postArchived'}typePostAgg={userId: string,content: string,archived: boolean}typePostCmd=|{type: 'createPost',userId: string,content: string}|{type: 'archivedPost',userId: string}constuser=createAggregate<UserEvt,UserAgg,'users'>({stream: 'users',create: ()=>({name: '',enabled: false}),fold: (evt)=>{switch(evt.type){case'created':
return{name: evt.name,enabled: true}case'disabled':
return{enabled: false}case'enabled':
return{enabled: true}}}})constpost=createAggregate<PostEvt,PostAgg,'posts'>({stream: 'posts',create: ()=>({content: '',userId: '',archived: false}),fold: (evt)=>{switch(evt.type){case'postCreated':
return{userId: evt.userId,content: evt.content}case'postArchived':
return{archived: true}},}})constprovider=createProvider()exportconst{ domain, createHandler }=createDomain({ provider },{ user, post })exportconstuserCmd=createCommands<UserEvt,UserEvt,UserCmd>(domain.user,{asynccreate(cmd,agg){ ... },asyncdisable(cmd,agg){ ... },asyncenable(cmd,agg){ ... },})exportconstpostCmd=createCommands<PostEvt,PostAgg,PostCmd>(domain.post,{asynccreatePost(cmd,agg){if(agg.version)thrownewCommandError('Post already exists')constuser=awaitdomain.user.getAggregate(cmd.userId)if(!user.version)thrownewCommandError('Unauthorized')return{type: 'postCreated',content: cmd.content,userId: cmd.userId}},asyncarchivePost(cmd,agg){if(cmd.userId!==agg.userId)thrownewCommandError('Not allowed')if(agg.archived)returnreturn{type: 'postArchived'}}})constpostModel=createHandler('posts-model',['posts'],{// When the event handler is started for the first time, the handler will begin at the end of the stream(s) historytailStream: false,// Every time the event handler is started, the handler will begin at the end of the stream(s) historyalwaysTailStream: false,// Skip events that throw an error when being handledcontinueOnError: false,})postModel.handle('posts','postCreated',async(id,event,meta)=>{// Insert into database})postModel.start()See API