I wanted to investigate how to implement something similar to my previous Event Sourcing examples (see Java and Clojure) but this time using Datomic. In particular, I wanted to keep the way commands are implemented in the domain, the way commands could be executed asynchronously and aggregates for consistency. If this sounds interesting I’m giving a talk at JavaZone with more details.

All example code for the blog post is available at rock-paper-scissors-with-datomic.

Domain code

I will use the exact same commands as my Event Sourced example.

(defrecord CreateGameCommand [aggregate-id player move]) (defrecord DecideMoveCommand [aggregate-id player move])

The protocol for handling commands will the same with the only difference is that it should return tx-data that will be passed to the transact function.

(defprotocol CommandHandler (perform [command state]))

We also need to define the schema for how to represent the game:

Game entity (state: GameState, created-by: Player, winner: Player, loser: Player, moves: Set[Move])

Player entity (email: String)

Move entity (player: Player, type: MoveType)

MoveType enum (rock, paper, scissors)

GameState enum (started, won, tied)

Both Game and Player are Aggregate roots. Move must always be a part of a Game. Entities obviously have a unique identifier. In addition the framework adds a version attribute to the aggregate roots for optimistic concurrency. A couple of things to note when looking at the actual Datomic schema:

Datomic uses the convention of prefixing attribute names with the entity it belongs to, for example :game/state (see db/ident)

(see db/ident) Datomic enums are named entities (see Making enum entity values)

Datomic components to support aggregates is a fairly recent addition. In this sample I have only enabled the feature without actually using it.

Lets see how the domain code turns out:

(extend-protocol CommandHandler CreateGameCommand (perform [command state] (when (:game/state state) (throw (Exception. "Already in started"))) [{:db/id #db/id[:db.part/user -1] :move/player (:player command) :move/type (:move command)} {:db/id (:aggregate-id command) :game/moves #db/id[:db.part/user -1] :game/state :game.state/started :game/created-by (:player command)}]) DecideMoveCommand (perform [command state] (when-not (= (:game/state state) :game.state/started) (throw (Exception. "Incorrect state"))) (when (= (:db/id (:game/created-by state)) (:player command)) (throw (Exception. "Cannot play against yourself"))) (let [creator-move (:move/type (first (:game/moves state))) creator-id (:db/id (:game/created-by state))] [{:db/id #db/id[:db.part/user -1] :move/player (:player command) :move/type (:move command)} (merge {:db/id (:aggregate-id command) :game/moves #db/id[:db.part/user -1]} (case (compare-moves (:move command) creator-move) :victory {:game/state :game.state/won :game/winner (:player command) :game/loser creator-id} :loss {:game/state :game.state/won :game/winner creator-id :game/loser (:player command)} :tie {:game/state :game.state/tied}))])))

Very similar to the event sourcing code! It looks like more code, but actually the main difference in lines of code is that I had to explicitly name all attributes and the Event Sourced version use the normal record constructor and is therefore much shorter. For comparison, here is the CreateGameCommand handler using the map-> construct with named parameters:

(extend-protocol CommandHandler CreateGameCommand (perform [command state] (when (:state state) (throw (Exception. "Already in started"))) [(map->GameCreatedEvent {:game-id (:aggregate-id command) :creator (:player command)}) (map->MoveDecidedEvent {:game-id (:aggregate-id command) :player (:player command) :move (:move command)})])

However, the most important difference is that the Datomic representation will be used for querying and the event sourced version is completely internal to the command processor. That is, Datomic is not segregating the command model from the query model! The most distinct example of this is how moves are represented:

In Datomic moves must be retrieved from another entity: creator-move (:move/type (first (:game/moves state)))

In the event sourced model we can just store the move directly on the game state: (:move state) . This might seem “dirty” as conceptually the game does not have a move, but it is very simple and since it is only used by the command code what is the harm?

Perhaps a better way to model is to introduce an attribute “last-move” on the game, and then both implementations could use this?

Infrastructure code

Since Datomic provides storage and querying all the infrastructure code needs to do is to provide consistency. But isn’t Datomic already ACID? What I actually mean is that if two commands for the same aggregate are executed at the same time then one of them should fail (and then perhaps be retried). One way to solve this is to move the execution of the command logic inside the transactor using a database functions. However, this would serialize the execution of ALL commands regardless of aggregate which definitely impacts scalability. I only need to serialize commands per aggregate.

I implemented optimistic locking using compare-and-set:

(defn handle-command [command conn] (let [aggregate-id (:aggregate-id command) state (datomic.api/entity (datomic.api/db conn) aggregate-id) modification (perform command state) old-version (:entity/version state) next-version (if (nil? old-version) 0 (inc old-version)) optimistic-concurrency [:db.fn/cas aggregate-id :entity/version old-version next-version] tx @(datomic.api/transact conn (conj modification optimistic-concurrency))] tx))

All that is needed is to append an invocation to the built-in cas function before calling transact and of course incrementing the version number.

I also added a method for generating entity ids which I need since the command contains the aggregate-id. I’m not at all happy with this implementation so please suggest improvements! Is there an idiomatic way in Datomic for generating ids client side?

(defn create-entity [conn] "Returns the id of the new entity." (let [optimistic-concurrency [:db.fn/cas #db/id[:db.part/user -1] :entity/version nil 0] tx @(datomic.api/transact conn [{:db/id #db/id[:db.part/user -1]} optimistic-concurrency])] (first (vals (:tempids tx)))))

Finally I need to setup the schema:

(defn initialize-schema [conn] (let [schema-tx (read-string (slurp "resources/schema.dtm"))] @(datomic.api/transact conn schema-tx)))

Test

Lets create an in-memory database and try some commands:

(def uri "datomic:mem://game") (datomic.api/create-database uri) (def conn (datomic.api/connect uri)) (initialize-schema conn) (def ply1 (create-entity conn)) (def ply2 (create-entity conn)) (def game-id (create-entity conn)) (handle-command (->CreateGameCommand game-id ply1 :move.type/rock) conn) (handle-command (->DecideMoveCommand game-id ply2 :move.type/scissors) conn) (println (datomic/touch (datomic/entity (datomic/db conn) game-id)))

And the result is (formatted by hand!):

{:db/id 17592186045428, :entity/version 2, :game/state :game.state/won, :game/created-by {:db/id 17592186045424}, :game/winner {:db/id 17592186045424}, :game/loser {:db/id 17592186045426}, :game/moves #{{:move/player {:db/id 17592186045426}, :move/type :move.type/scissors, :db/id 17592186045434} {:move/player {:db/id 17592186045424}, :move/type :move.type/rock, :db/id 17592186045432}}}

Notice how the touch function pulls in all moves since they are marked as a component.

I’m definitely happy with the results!

Conclusion

Using Datomic instead of Event Sourcing is very similar from the command side and the main differences are: