CQRS-EventStore

You need Node.js 4+ to use it!

Installation

$ npm install cqrs-eventstore

Working example

A full example is provided in the demo folder. To run it:

• cd to the demo folder
• npm install
• node event-store-demo.js

Domain and DTOs

In order to use CQRS-EventStore, you need to implement your own aggregate and DTOs. Your aggregate must extend Aggregate.

An aggregate example including the DTOs:

"use strict";classBaseEvent{constructor(){this.idthis.version}}module.exports=BaseEvent

"use strict"constBaseEvent=require("./baseEvent")classAddressUpdatedextendsBaseEvent{constructor(address){super()this.address=address}}module.exports=AddressUpdated

"use strict"constBaseEvent=require("./baseEvent")classMobileUpdatedextendsBaseEvent{constructor(mobile){super()this.mobile=mobile}}module.exports=MobileUpdated

"use strict"constBaseEvent=require("./baseEvent")classUserInfoCreatedextendsBaseEvent{constructor(name,surname,address,mobile){super()this.name=namethis.surname=surnamethis.address=addressthis.mobile=mobile}}module.exports=UserInfoCreated

"use strict"constNodeEventStore=require("cqrs-eventstore")constUserInfoCreated=require("./dto/userInfoCreated")constAddressUpdated=require("./dto/addressUpdated")constMobileUpdated=require("./dto/mobileUpdated")constclone=require("clone")//Clone is used for the snapshot, it's totally up to you how to implement it.functionUserInfo(id){//We are not exposing the UserInfo to the outside world, but we access to it through query.functionUserInfoObj(){this.namethis.surnamethis.addressthis.mobile}let_userInfoclassUserInfoextendsNodeEventStore.Aggregate{constructor(id){super(id)_userInfo=newUserInfoObj()}snapshot(){returnclone(_userInfo)}applySnapshot(payload){_userInfo=payload}//QueriesgetMobile(){return_userInfo.mobile}getAddress(){return_userInfo.address}//Mutatorsinitialize(name,surname,address,mobile){super.raiseEvent(newUserInfoCreated(name,surname,address,mobile))}updateAddress(address){super.raiseEvent(newAddressUpdated(address))}updateMobile(mobile,hookFn){super.raiseEvent(newMobileUpdated(mobile),hookFn)}//ApplyUserInfoCreated(payload){_userInfo.name=payload.name_userInfo.surname=payload.surname_userInfo.address=payload.address_userInfo.mobile=payload.mobile}AddressUpdated(payload){_userInfo.address=payload.address}MobileUpdated(payload){_userInfo.mobile=payload.mobile}}returnnewUserInfo(id)}module.exports=UserInfo

Implementing the persistence layer

In order to implement your own persistence layer, you need to extend PersistenceAdapter and register it into the configurator (I'll show it later). The methods save, readSnapshot and readEvents must be implemented. All methods must return a promise. In the save method you need to persist your events and snapshot.

Below an example how to implement a sqlite persistor.

"use strict"constnodeEventStore=require("cqrs-eventstore")constfs=require("fs");constsqlite3=require("sqlite3").verbose();const_=require("underscore")constutil=require("util")constuuid=require("uuid")classSqlitePersistorextendsnodeEventStore.PersistenceAdapter{constructor(){super()constfile="eventStore.db";constexists=fs.existsSync(file);this.db=newsqlite3.Database(file);this.db.serialize(()=>{if(!exists){this.db.run("CREATE TABLE Events (id TEXT, streamId TEXT, version INTEGER, timestamp TEXT, eventType TEXT, payload BLOB)");this.db.run("CREATE TABLE Snapshots (id TEXT, streamId TEXT, version INTEGER, timestamp TEXT, payload BLOB)");}});}save(events,snapshots){constself=this;returnnewPromise((resolve,reject)=>{self.db.serialize(()=>{try{self.db.run("BEGIN TRANSACTION")_.each(events,(e)=>{self.db.run("INSERT INTO Events VALUES (?, ?, ?, ?, ?, ?)",uuid.v4(),e.streamId,e.version,newDate(),e.eventType,e.payload)})_.each(snapshots,(e)=>{self.db.run("INSERT INTO Snapshots VALUES (?, ?, ?, ?, ?)",uuid.v4(),e.streamId,e.version,newDate(),e.payload)})self.db.run("COMMIT TRANSACTION")resolve()}catch(err){self.db.run("ROLLBACK TRANSACTION")reject(err)}})})}//return a promisereadSnapshot(id){returnnewPromise((resolve,reject)=>{this.db.get("SELECT * FROM Snapshots WHERE streamId = ? ORDER BY version DESC LIMIT 1",[id],(err,row)=>{if(err)returnreject(err)resolve(row)});})}//return a promisereadEvents(id,fromVersion){returnnewPromise((resolve,reject)=>{this.db.all("SELECT * FROM Events WHERE streamId = ? AND version > ? ORDER BY version",[id,fromVersion],(err,rows)=>{if(err)returnreject(err)resolve(rows)});})}}module.exports=newSqlitePersistor()

Implementing hooks

CQRS-EventStore comes with a build-in hook functionality. We can execute a task after each commands.

A simple hook that print into the console on each mobile number update:

"use strict"constutil=require("util")module.exports=evt=>{console.log(util.format("Mobile number updated %s",evt.mobile))}

Hooks need to be registered into the configurator

Configuration

Before to use CQRS-EventStore, we need to configure it.

The parameters are:

• cacheExpiration: cache expiration in seconds, the default is 0 (unlimited).
• cacheDeleteCheckInterval: The period in seconds, used for the automatic delete check interval. Default is 60 seconds.
• repository: your extended persistance layer, if not provived an in-memory persistence will be used.
• snapshotEvery: event threshold for the snapshot, the default is 0 (snapshot disabled). For example, if we assign 50, every 50 events we create the snapshot.
• payloadSerializationFormat: payload serialization/compression, default is NodeEventStore.serializationFormats.stringify
  The "cqrs-eventstore" module exposes an 'enumeration' called serializationFormats. Available values are: stringify, zip, unserialized.

Usage Example

"use strict"constNodeEventStore=require("cqrs-eventstore")constUserInfoAggregate=require("./userInfoAggregate")constmobileUpdatedHook=require("./mobile-updated-hook")//We need to register the hooks here, the name of the hook must match the apply methodNodeEventStore.registerHook("MobileUpdated",mobileUpdatedHook)//ConfigurationconstEventStore=NodeEventStore.initialize({cacheExpiration: 180,cacheDeleteCheckInterval: 60,repository: require("./sqlite-persistor"),snapshotEvery: 5,payloadSerializationFormat: NodeEventStore.serializationFormats.zip})constrepository=newEventStore.Repository(UserInfoAggregate)letuserInfoAggregate=newUserInfoAggregate(1)userInfoAggregate.initialize("Gennaro","Del Sorbo","Main Street","09762847")repository.save(userInfoAggregate).then(()=>{userInfoAggregate.updateMobile("333");userInfoAggregate.updateMobile("334");userInfoAggregate.updateMobile("335");userInfoAggregate.updateAddress("12, Main St.")userInfoAggregate.updateAddress("15, Main St.")repository.save(userInfoAggregate).then(()=>{console.log("all saved")console.log("try a read")repository.read(1).then(userInfo=>{console.log(userInfo.Mobile)console.log(userInfo.Address)}).catch(e=>{console.log(e)});})}).catch(err=>{console.log(err)})

Contributing

1. clone this repo

2. npm run setup

Run the demo

npm start

Run the tests

npm test

Run the tests and listen for a debugger

npm run test-debug