lukaszb / cq Goto Github PK

Currently those are Django model instances. They should be translated in and out by the storage engine.

It's actually confusing right now. We are using metaclass and specially prepared Action class that should be used to indicate a domain actions. This is unnecessary convoluted - we should fall back to simple strings when identifying domain events. In addition, this would help to handle events coming from external services (as they might not have action classes at all). We currently support both Action class and strings.

Reference: https://github.com/lukaszb/ses/blob/0580d003067c0f559bdff4fcdb1d7394acefc535/ses/entities.py#L1

One way we can accomplish that is to add version field to events and make unique constraint on 2 fields that identify entity at particular moment in time (entity's id and version).

There are two possible outcomes:

1. one write would fail, leaving up to the client to decide if it want's to retry
2. both would succeed, however this would make the client to need to re-fetch the entity to make sure it's state is in a desired shape

We are going to implement 1) approach with a custom exception.

I believe that cq should be framework/tools agnostic, however I caught myself couple of times running from cq.models import Event within my Django based app.

We should be able to query events in a much better manner. To be specific I would love to be able to query events with explicit ordering and filtering after doing simple library import.

Basically, I'd like to achieve something like:

import cq

for event in cq.query_events('User.StartedSession', limit=50):
    print("{} | User {} logged in".format(event.ts, event.data['email']))

Or something like:

print(cq.query_events(ts="3 days")

which should show events from last 3 days.

Signature of cq.query_events

def query_events(*names, limit=100, order_by='-ts', **filters):
    ...

• names are not named arguments. Might be empty of course - in this case all events, regardless of their types, would be returned
• limit results to first 100 by default. If None given, all matching events are returned
• order_by defaults to -ts, meaning events would be ordered by date of creation, descending. Allows to change or disable order.
• filters: For now it would probably suffice to filter by ts param. It should be similar to what is possible with Django ORM: ts=datetime(2017, 3, 24) (exact comparison) or ts__gte=datetime(2017, 3, 24) (all events created after 2017-03-24) or ts="day" (if string given, should be human readable and would be treated as "show me events from last day/week/month" or number of those like "2 weeks". I'd say this is optional, however I'd strongly advice to implement it as it would be mostly used in my opinion).

Should be done in "implement usage first".

Let's imagine we have an User.Registered event with email and password. After a while we decide that whenever we register an user we also want to apply a role for her. In such case we should upgrade a command that generates the event but then we would end up with old and new events that would have different set of data. To prevent that, we would simply recognise old events and upcast them to new version. This could looks something like:

cq.register_upcasters([
    cq.upcaster(event='User.Registered', revision=1, method='accounts.upcasters.add_role'),
])

def add_role(event):
    event.data['role'] = 'DEFAULT_USER'

We might use classes for upcasting too, like:

cq.register_upcasters([
    accounts.upcasters.AddRole',
])


class AddRole(cq.Upcaster):
    aggregate_type = 'User'
    name = 'Register'
    apply_on_revision = 1

    def add_role(self, event):
        event.data['role'] = 'DEFAULT_USER'

Even if we would be working with entities most of the time probably it doesn't harm if we call them aggregates as this is more generic object. In this task we would also need to update all variable names.

Before writing any more code we should try to explain to users how to use this library.

We'd need to re-read it and make sure it's simple enough.

now they only accept data but sometimes it would be necessary to for example take event's timestamp into account

Not sure anyone will even touch this without one. Suggest BSD or MIT.

https://github.com/lukaszb/cq/blob/devel/cq/storages.py#L66

Signature of Storage.get_events should be extended with required aggregate_type param. Now, if aggregate id clashes with another aggregate type, an error would be thrown (there would be no mutator for found aggregate).

We need a method to re-run all events.

There are couple of gotchas:

• what to do with existing projections? (I'd say we need a way to cleanup all projections first)
• how to treat sagas (commands that spawns other commands) ? We can probably simply omit sagas as purpose of them are simply to create events by calling other commands. We'd need a way to mark a command a saga, though.
• how about commands for which handlers are running not-idempotent action? Like sending an e-mail. Let's say we have a register_user command and it's handler not only updates user model projection but also sends out an e-mail with an activation link. We should make sure that running rehydration process won't invoke such handler actions (and still updates projections)

This should be done just after public release but initial branch with travis config should be done before that.

Currently Repository.get_aggregate method is not very efficient - it fetches all events that were used to create an aggregate. In order not to do that we want to introduce snapshots.

Change how aggregate is retrieved

Logic behind the get_aggregate should be changed as follow:

• fetch snapshot with greatest version number for particular aggregate (or all events if there are no snapshot yet)
• fetch all events with version greater than that from the snapshot

Store snapshots

• Add Snapshot model
• Add Storage.create_snapshot method
• Create snapshot "every now and then" (i.e. every 20 events - this variable should be controlled somehow)