Building Microservices Through Event-Driven Architecture, Part 7: Implementing Event Sourcing on Repositories

the repository is responsible for adding events to the eventstore and also retrieving all events from the event store.

when the aggregate is saved, then all uncommitted events related to that aggregateroot are added to the eventstore table.

the eventstore table is append-only (update and delete are not allowed).

you may also like: building microservices through event-driven architecture, part 1: application-specific business rules

the schema of the eventstore table will look like this:

• the id is the primary key.
• version is the version of the aggregate.
• aggregateid is the identifier of the aggregate.
• name is the name of the event: 2@735f8407-16be-44b5-be96-2bab582b5298.
• typename is the type of the event: logcorner.edusync.speech.domain.events.speech.speechcreatedevent, logcorner.edusync.speech.domain, version=1.0.0.0, culture=neutral, publickeytoken=null
• occurredon is the event date.
• the payload is the event stream.

{

"title": { "value": "introducing azure cosmos db" }, 

"url": { "value": "https://azure.microsoft.com/en-us/resomurcejjjnns/videos/azure-friday-introducing-azurkke-cosmos-db_g5/" }, 

"description": { "value": "kirill gavrylyuk stops by azure friday to talk cosmos db with scott hanselman. watch quick overview of the industry's first globally distributed multi-model database service followed by a demo of moving an existing mongodb app to cosmos db with a single config change." }, 

"type": { "value": 2 }, 

"aggregateid": "735f8407-16be-44b5-be96-2bab582b5298", 

"eventid": "6eb58cb4-da5e-46d4-8325-e742a20935ab", 

"aggregateversion": 0, 

"ocurrendon": "2019-09-08t10:55:48.5528117z" 

}

event-sourcing interfaces

let us define ieventstorerepository interfaces

event-sourcing implementation

let us define the eventstore class.

• the id is the identifier of the event stream.
• version is the current aggregate version.
• aggregateid is the identifier of the aggregate.
• name is the name of the event stream.
• typename is the full type of the event (ex : logcorner.edusync.speech.domain.events.speech.speechcreatedevent, logcorner.edusync.speech.domain, version=1.0.0.0, culture=neutral, publickeytoken=null)
• occurredon is the date on which the event occurred.
• serializedbody is the event serialized in json.
• appendasync is a function that indicates whether the event is synchronized or not yet.

appendasync implementation

so the first test is appendasync should append an event on eventstore

test case 1: appendasync should append an event on eventstore

here i have to mock a context , create an instance of ieventstorerepository and call appendasync with an eventstore object, then the context should have a set of eventstore with a single element and that element should be equals to the eventstore object i passed in the argument of appendasync.

the test may look like this:

instead of using moq i will use entityframeworkcore in-memory database which is a simpler way of database unit testing.

the next step is to create an eventstorerepository class.

then add a dbset<eventstore> eventstore property to databasecontex class.

the code compiles but test fails because dbset<eventstore> eventstore is null.

let us config some mappings.

eventstoreentitytypeconfiguration class has reponsibility to map eventstore class to eventstore database table.

the final test looks like this:

getbyidasync implementation

getbyidasync retrieves all events related to the aggregate from the event store.

here i have to create an empty aggregate of type speech, read all events from the eventstore related to it and finally apply the events.

test case 2: createinstance of aggregateroot should return an empty aggregate

let us create an invoker class with the responsibility to create an empty aggregateroot.

the test pass.

test case 3: getbyidasync with badaggregateid should raise badaggregateidexception

given an bad aggregateid (ex: empty aggregateid), getbyidasync should raise an exception (badaggregateidexception).

test case 4: getbyidasyncwithnullinstanceofaggregateshouldraisenullinstanceofaggregateidexception

given an aggregateid that does not exist (ex: empty aggregateid), getbyidasync should raise an exception (nullinstanceofaggregateidexception).

test case 5: getbyidasyncwithouteventsshouldreturnemptylist

if there is no event related to a given aggregateid, then getbyidasyncshould returns an empty list.

test case 6: getbyidasyncwitheventsshouldreturnthecurrentstateofaggregate

here, i have to read all events related to the given aggregateid from the event store, rebuild the aggregate state and return it.

to reach this goal, i have to deserialize the event which is in json structure according to the event type.

test case 6.1: deserialize event stream should return an event

here i will create a function that deserializes a json string to an event object.

test case 6.2: getbyidasyncwitheventsshouldreturnthecurrentstateoftheaggregate

here, i will finish the implementation of my function.

the source code of this article is available on github (feature/task/ eventsourcingrepository ).

regards!

further reading