eventstore.clj

(defprotocol Command
  (perform [command state] "Either throw an exception to abort, or return a sequence of resulting events to be applied to the state."))

(defprotocol Event
  (affect [this state] "Derive a new state from the given state"))

(defprotocol EventStore
  (event-seq [this] "Returns a seq of all events in the store")
  (append [this events] "Append events to the store"))

(defn apply-events
  "Derive a new state from a sequence of events"
  [state events]
  (reduce #(affect %2 %1) state events))

(defrecord DB [state-ref event-store-agent])

(defn state [db] @(:state-ref db))

(defn command!
  "Submit a command to be run and stored against the db"
  [db command]
  (dosync
   (let [state-ref (:state-ref db)
         new-events (perform command @state-ref)]
     (alter state-ref apply-events new-events)
     (send-off (:event-store-agent db) append new-events)
     @state-ref)))

;; implementations

(defn- transaction-seq
  "Return a seq of transactions (forms) in the data file"
  [path]
  (with-open [r (java.io.PushbackReader. (io/reader path))]
    (doall
     (take-while #(not= ::EOF %)
                 (repeatedly (fn []
                               (try (read r)
                                    (catch RuntimeException e
                                      ::EOF))))))))

(defrecord JournalFileEventStore [path]
  EventStore
  (event-seq [this]
    (io!
     (when (.exists (io/file path))
       (transaction-seq path))))
  (append [this event]
    (io!
     (let [output (str ";; " (java.util.Date.) "\n" (pr-str event) "\n")]
       (spit path output :append true)
       (->JournalFileEventStore path)))))

(defn journal-db
  "Returns a DB backed by a journal file"
  [path]
  (let [event-store (->JournalFileEventStore path)
        init-state (apply-events {} (flatten (event-seq event-store)))]
    (->DB (ref init-state) (agent event-store))))