rill.wheel — rill-event-sourcing/wheel 0.1.24

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 "[e-mail]") registered :registered?)
  => true

(registered-event (user "[e-mail]"))
  => {:rill.message/type :user/registered,
      :email "[e-mail]",
      :rill.wheel/type :user/user}

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

Store and retrieve aggregates in a repository

(-> (get-user repo "[e-mail])
    (registered)
    (command/commit!))
;; ...
(get-user some-repository "[e-mail]")

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

aggregate clj

(aggregate result)

Return the aggregate from a result. If the result is an aggregate, returns it as is.

aggregate? clj

(aggregate? obj)

Test that obj is an aggregate.

apply-command clj

(apply-command aggregate command)

Given a command and aggregate, apply the command to the aggregate. Should return an updated aggregate or a rejection.

apply-command* clj multimethod

apply-event clj

(apply-event aggregate event)

Update the properties of aggregate given event. Implementation for different event types will be given by defevent.

apply-event* clj multimethod

apply-new-event clj

(apply-new-event aggregate event)

Apply a new event to the aggregate. The new events will be committed when the aggregate is committed to a repository.

apply-stored-event clj

(apply-stored-event aggregate event)

Apply a previously committed event to the aggregate. This increments the version of the aggregate.

command-result? clj

(command-result? r)

commit! clj

(commit! aggregate-or-rejection)

Commit the result of a command execution. If the command returned a rejection nothing is committed and the rejection is returned. If the result is an aggregate it is committed to the repository. If that succeeds an ok is returned. Otherwise a conflict is returned.

conflict clj

(conflict aggregate)

Creates a conflict. Use conflict when there were changes to the aggregate in the repository in the time between fetching the aggregate and calling commit!.

conflict? clj

(conflict? result)

Was there a conflict between fetching the aggregate and calling commit?

defaggregate clj macro

(defaggregate name
              doc-string?
              attr-map?
              [properties*]
              pre-post-map?
              events?
              commands?)

Defines an aggregate type, and aggregate-id function. The aggregate's id key is a map with a key for every property in properties*, plus the aggregate type, a qualified keyword from name.

Also defines a function get-{name}, which takes an additional first repository argument and retrieves the aggregate.

events? and commands? are sequences of event specs and command specs and passed to defevent and rill.wheel/defcommand respectively.

defcommand clj macro

(defcommand name
            aggregate-type
            doc-string?
            attr-map?
            [repository properties*]
            pre-post-map?
            body)

Defines a command as a named function that takes any arguments and returns an aggregate or rejection that can be passed to commit!.

The metadata of the command may contain a :rill.wheel/events key, which will specify the types of the events that may be generated. As a convenience, the corresponding event functions are declared automatically so the defevent statements can be written after the command. This usually reads a bit nicer.

 (defcommand answer-question ::question
   "Try to answer a question"
   [question user-id answer]
   (if (some-check question answer)
     (answered-correctly question user-id answer)
     (answered-incorrectly question user-id answer)))

defevent clj macro

(defevent name
          aggregate-type
          doc-string?
          attr-map?
          [aggregate properties*]
          pre-post-map?
          body*)

Defines function that takes aggregate + properties, constructs an event and applies the event as a new event to aggregate. Properties defined on the aggregate-type definition will be merged with the event; do not define properties with defevent that are already defined in the corresponding defaggregate.

For cases where you only need the event and can ignore the aggregate, the function "{name}-event" is defined with the same signature. This function is used by the "{name}" function to generate the event before calling apply-event (see below).

The given prepost-map, if supplied gets included in the definition of the "{name}-event" function.

The given body, if supplied, defines an apply-event multimethod that applies the event to the aggregate. If no body is supplied, the default apply-event will be used, which will return the aggregate as is.

empty clj

(empty aggregate-id)

Create a new aggregate with id aggregate-id and no events. Aggregate version will be -1. Note that empty aggregates cannot be stored.

empty? clj

(empty? aggregate)

Test that the aggregate is new and has no uncommitted events.

exists clj

(exists aggregate)

If aggregate is not new, return aggregate, otherwise nil.

fetch-aggregate clj multimethod

Given a command and repository, fetch the target aggregate.

merge-aggregate-props clj

(merge-aggregate-props aggregate partial-event)

message? clj

(message? m)

new-events clj

(new-events aggregate)

The events that will be committed when this aggregate is committed.

new? clj

(new? aggregate)

Test that the aggregate has no committed events.

ok clj

(ok aggregate)

Creates an ok? object describing the committed events and aggregate.

ok? clj

(ok? result)

Is result an ok? object?

reason clj

(reason rejection)

Return the reason for a rejection. Returns :rill.wheel/conflict for a conflict.

rejection clj

(rejection aggregate reason)

Create a rejection for aggregate with reason.

rejection? clj

(rejection? result)

Checks if result is rejected.

repository clj

(repository aggregate)

Return the repository of aggregate.

transact! clj

(transact! repo command)

Run and commit the given command against the repository.

type clj

(type aggregate)

Return the type of this aggregate.

type-properties clj

(type-properties t)

The properties of the identifier of aggregate type t.

uncommitted? clj

(uncommitted? aggregate)

aggregate has events applied that can be committed.