Grain Read Model Pattern

Required Imports

(ns my.service.core.read-models
  (:require [ai.obney.grain.event-store-v3.interface :as event-store]))

Event Type Sets

Define which events affect each read model:

(def contact-events
  #{:crm/contact-created
    :crm/contact-updated
    :crm/contact-field-set
    :crm/contact-archived
    :crm/contact-deleted})

(def relationship-events
  #{:crm/relationship-created
    :crm/relationship-updated
    :crm/relationship-ended})

Projection Multimethod

;; Dispatch on event type
(defmulti contacts*
  (fn [_state event] (:event/type event)))

;; Handle creation
(defmethod contacts* :crm/contact-created
  [state event]
  (assoc state (:contact-id event)
         {:id (:contact-id event)
          :type-slug (:type-slug event)
          :display-name (:display-name event)
          :field-values (:field-values event)
          :status :active
          :created-at (str (:event/timestamp event))}))

;; Handle update
(defmethod contacts* :crm/contact-updated
  [state event]
  (-> state
      (update (:contact-id event) merge (:changes event))
      (assoc-in [(:contact-id event) :updated-at] (str (:event/timestamp event)))))

;; Handle field set
(defmethod contacts* :crm/contact-field-set
  [state event]
  (-> state
      (assoc-in [(:contact-id event) :field-values (keyword (:field-slug event))]
                (:new-value event))
      (assoc-in [(:contact-id event) :updated-at] (str (:event/timestamp event)))))

;; Handle archive
(defmethod contacts* :crm/contact-archived
  [state event]
  (assoc-in state [(:contact-id event) :status] :archived))

;; Default: ignore unknown events
(defmethod contacts* :default [state _] state)

;; Reduction function
(defn contacts
  "Build contacts read model from events"
  [initial-state events]
  (reduce contacts* initial-state events))

Query Helper Functions

Get Single Entity

(defn get-contact
  "Get a single contact by ID"
  [event-store contact-id]
  (let [events (event-store/read event-store
                                 {:types contact-events
                                  :tags #{[:contact contact-id]}})]
    (get (contacts {} events) contact-id)))

Get All Entities

(defn get-contacts-all
  "Get all contacts"
  [event-store]
  (let [events (event-store/read event-store {:types contact-events})]
    (vals (contacts {} events))))

Get by Filter

(defn get-contacts-by-type
  "Get contacts of a specific type"
  [event-store type-slug]
  (let [events (event-store/read event-store {:types contact-events})]
    (->> (contacts {} events)
         vals
         (filter #(= type-slug (:type-slug %))))))

Batch Loading

(defn get-contacts-batch
  "Get multiple contacts by IDs in a single read"
  [event-store contact-ids]
  (when (seq contact-ids)
    (let [tags (set (map #(vector :contact %) contact-ids))
          events (event-store/read event-store
                                   {:types contact-events
                                    :tags tags})]
      (contacts {} events))))

Composing Projections

(defn get-contact-with-attribution
  "Get contact with attribution in single query"
  [event-store contact-id]
  (let [events (event-store/read event-store
                                 {:types (into contact-events attribution-events)
                                  :tags #{[:contact contact-id]}})
        contact (get (contacts {} events) contact-id)
        attr (get (attributions {} events) contact-id)]
    (cond-> contact
      attr (assoc :attribution attr))))

Event Store Read Options

;; By event types
{:types #{:crm/contact-created :crm/contact-updated}}

;; By tags (entities)
{:tags #{[:contact contact-id]}}

;; After specific event (for incremental updates)
{:types #{:crm/contact-field-set}
 :after last-processed-event-id}

;; Combined
{:types contact-events
 :tags #{[:contact contact-id]}
 :after last-event-id}

Conventions

• State is a map keyed by entity ID: {uuid1 {...} uuid2 {...}}
• Use assoc for create, update/assoc-in for modify
• Always provide :default method to ignore unknown events
• Query helpers read events then reduce to current state
• Use tags for efficient filtering (avoid reading all events)
• Batch load related data to avoid N+1 queries

Reference Files

• components/user-service/src/ai/obney/workshop/user_service/core/read_models.clj - User projections
• components/crm-service/src/ai/obney/workshop/crm_service/core/read_models.clj - CRM projections