Aggregates and Events

Synopsis

(require '[rill.wheel :as aggregate
           :refer [defaggregate defevent]])

(defaggregate user
  "a user is identified by a single `email` property"
  [email])

(defevent registered ::user
  "user was correctly registered"
  [user]
  (assoc user :registered? true))

(defevent unregistered ::user
  "user has unregistered"
  [user]
  (dissoc user :registered?))

(-> (user "user@example.com") registered :registered?)
  => true

(registered-event (user "user@example.com"))
  => {:rill.message/type :user/registered,
      :email "user@example.com",
      :rill.wheel/type :user/user}

(wheel/new-events some-aggreate)
  => seq-of-events

Store and retrieve aggregates in a repository

(-> (get-user repo "user@example.com)
    (registered)
    (command/commit!))
;; ...
(get-user some-repository "user@example.com")

Full example of defaggregate

(defaggregate turnstile
  "An aggregate with docstring"
  [turnstile-id]
  {:pre [(instance? java.util.UUID turnstile-id)]}
  ((installed
    "A turnstile was installed"
    [turnstile]
    (assoc turnstile
           :installed? true
           :locked? true
           :coins 0
           :turns 0
           :pushes 0))

   (coin-inserted
    "A Coin was inserted into the turnstile"
    [turnstile]
    (-> turnstile
        (update :coins inc)
        (assoc :locked? false)))

   (arm-turned
    "The turnstile's arm was turned"
    [turnstile]
    (-> turnstile
        (update :pushes inc)
        (update :turns inc)
        (assoc :locked? true)))

   (arm-pushed-ineffectively
    "The arm was pushed but did not move"
    [turnstile]
    (-> turnstile
        (update :pushes inc))))

  ((install-turnstile
    [repo turnstile-id]
    (let [turnstile (get-turnstile repo turnstile-id)]
      (if (wheel/exists turnstile)
        (rejection turnstile "Already exists")
        (installed turnstile))))

   (insert-coin
    "Insert coin into turnstile, will unlock"
    [repo turnstile-id]
    (let [turnstile (get-turnstile repo turnstile-id)]
      (if (:installed? turnstile)
        (coin-inserted turnstile)
        (rejection turnstile "Turnstile not installed"))))

   (push-arm
    "Push the arm, might turn or be ineffective"
    {::wheel/events [::arm-pushed-ineffectively ::arm-turned]}
    [repo turnstile-id]
    (let [turnstile (get-turnstile repo turnstile-id)]
      (cond
        (not (:installed? turnstile))
        (rejection turnstile "Not installed")
        (:locked? turnstile)
        (arm-pushed-ineffectively turnstile)
        :else
         (arm-turned turnstile))))))

Commands

Commands are functions that apply new events to aggregates.

Command flow

 (-> (get-some-aggregate repository id) ; 1.
     (cmd-call additional-argument)     ; 2.
     (commit!))                         ; 3.

1. Fetch aggregate

Before calling the command, the aggregate it applies to should get fetched from the repository. In rill/wheel, this will always work and must be done even for aggregates that have no events applied to them - this will result in an rill.wheel/empty? aggregate that can be committed later.

2. Calling the command

A command can have any number of arguments, and it’s idiomatic for commands to take the aggregate-to-change as the first argument.

As a matter of style, it’s suggested that commands do not fetch other objects from the repository but are explicitly passed any necessary aggregates.

Success

A successful command returns an uncommitted? aggregate.

Rejection

A command may be rejected, in which case the command returns a rejection - meaning the request was denied for business reasons. Rejections are explicitly constructed in the defcommand body by the application writer.

It’s typically useless to retry a rejected command.

3. Committing results

The result of a command can be persisted back to the repository by calling commit!. If commit! is passed a rejection it will return it. Otherwise the argument should be an aggregate that will be persisted.

ok

A successful commit returns an ok? object describing the committed events and aggregate.

conflict

Committing an updated aggregate can return a conflict, meaning there were changes to the aggregate in the repository in the time between fetching the aggregate and calling commit!.

Depending on the use case, it may be useful to update the aggregate and retry a conflicted command.

Defining commands

 (defaggregate x                                 ; 1.
   [id])

 (defevent x-happened                            ; 2.
    "X happened to obj"
    [obj arg1]
    (assoc obj :a arg1))

 (defcommand do-x ::x                            ; 3.
    "Make X happen to obj"
    [obj arg1]
    (if (= (:a obj) arg1))                       ; 4.
        (rejection obj "Arg already applied")
        (x-happened obj)))                       ; 5.

1. Define the aggregate type

2. Define events to apply

Commands can only affect aggregates by applying events to them. Here we define an event with defevent. When the x-happened event is applied it will set key :a of aggregate obj.

It’s idiomatic to give events a past-tense name, to indicate that the event happened and cannot be rejected.

3. Define command

Commands are defined by calling defcommand, specifying a name, aggregate type, optional docstring, argument vector and a command body.

4. Test state and reject command

Aggregate state is typically only used to keep track of information that must be used to validate commands. When a command must not proceed, the command body can return a rejection with a reason.

5. Apply new event(s)

When the command is acceptable, it should apply the necessary events to the aggregate and return the result (the updated aggregate).

Alternative command invocations

defcommand installs a number of functions and multi-methods that can be used to invoke the defined command.

In the following examples, we assume the following definitions:

 (ns user
    (:require [rill.wheel :as aggregate
                          :refer [defaggregate
                                  defcommand
                                  defevent
                                  transact!
                                  ok?]]))

 (defaggregate user
    [user-id])

 (defevent registered ::user
    [user name])

 (defcommand register ::user
    [user name]
    (registered user name))

Commands as data

If you need to, you can describe and execute any command defined with defcommand as a message. The commands are implemented as maps with a rill.message/type key indicating the command type with a qualified keyword.

Command constructor

For every (defcommand cmd-name ...) definition, a constructor function named ->cmd-name is created that will create a valid rill.wheel command message:

(->register "my-id" "Some Name")

=> {:rill.message/type :user/register,
    :user-id "my-id",
    :name "Some Name"}

Note that the ->register function also takes the identifying properties of the user aggregate. This is required so that the correct user aggregate can be fetched from the repository.

transact!

You can run the given command message directly against the repository using transact!. This will fetch the aggregate using the fetch-aggregate function, calls apply-command and commit! the result.

(ok? (transact! repository
                (->register "my-id"
                            "Some Name")))

=> true

If your commands are invoked from some remote source (like a single-page application - see the mpare-net/weir project), these are the semantics you probably want (excluding authentication etc).

apply-command

The apply-command function is used by transact! and takes the aggregate to update and the command message and executes the defcommand body:

 (-> (get-user repo my-id)
     (apply-command (->register my-id my-name)))

fetch-aggregate

Also used by transact!, this multi-method takes the repository and the command message and returns the aggregate the command should be applied to.

Commands as functions

It’s convenient to be able to call commands directly as regular names functions. For this purpose there are two flavors:

command-name!

A generated function named after the command with an exclamation mark added. Takes the repository and all properties to identify the aggregate plus additional command properties, and commit! the result.

 (ok? (register! repository "user-id" "Some Name"))

 => true

command-name

Another generated function that takes the aggregate as the first argument and the additional command arguments and applies the command to the aggregate. This does not commit but returns the updated aggregate or a rejection. You can chain successful calls to named command functions and commit! the result.

 (ok? (-> repository
        (get-user "user-id")
        (register "Some Name")
        (commit!)))

 => true

See also

• rill.event-store