Grain Todo Processor Pattern

Todo processors are policies that react to domain events and emit follow-up events or perform side effects.

Required Imports

(ns ai.obney.orc.my-service.core.todo-processors
  (:require [ai.obney.orc.my-service.core.read-models :as rm]
            [ai.obney.grain.event-store-v3.interface :refer [->event]]
            [ai.obney.grain.todo-processor-v2.interface :refer [defprocessor]]
            [ai.obney.grain.time.interface :as time]))

Processor Template

(defprocessor :ns processor-name
  {:topics #{:ns/triggering-event}}
  "Docstring explaining what this processor does and when."
  [context]
  (let [event (:event context)
        entity-id (:entity-id event)
        ;; Check current state using read models
        entity (rm/get-entity context entity-id)]
    (if (should-take-action? entity)
      ;; Emit follow-up events
      {:result/events
       [(->event
         {:type :ns/follow-up-action
          :tags #{[:entity entity-id]}
          :body {:entity-id entity-id
                 :triggered-by (:event/type event)
                 :processed-at (str (time/now))}})]}
      ;; No action needed
      {})))

The generated var name is ns-processor-name (e.g., (defprocessor :crm ensure-attribution ...) creates var crm-ensure-attribution).

Return Patterns

Emit Events (pure -- batch checkpointed)

{:result/events [event1 event2]}

Side Effect with Checkpoint Control

{:result/effect (fn [] (send-email! ...))
 :result/checkpoint :after}  ;; or :before

No Action

{}

Auto-Registration

defprocessor registers the processor in a global registry automatically. There is no manual registry map (def todo-processors) needed. The control plane discovers processors from the registry and starts them when a tenant is assigned to the node.

Interface.clj only needs a bare side-effect require to load the namespace:

(ns ai.obney.orc.my-service.interface
  (:require ;; Side-effect requires
            [ai.obney.orc.my-service.core.commands]
            [ai.obney.orc.my-service.core.queries]
            [ai.obney.orc.my-service.core.todo-processors]  ;; bare require, no alias
            ...))

;; No todo-processors re-export needed

No manual wiring is needed -- the control plane discovers processors from the global registry.

Common Use Cases

Ensure Default Value

(defprocessor :crm ensure-attribution
  {:topics #{:crm/contact-created}}
  "Record default attribution if none exists."
  [context]
  (let [{:keys [contact-id]} (:event context)
        existing (rm/get-contact context contact-id)]
    (if (:attribution existing)
      {}  ;; Already has attribution
      {:result/events
       [(->event
         {:type :crm/attribution-recorded
          :tags #{[:contact contact-id]}
          :body {:contact-id contact-id
                 :attribution {:source :manual_entry
                               :recorded-at (str (time/now))}}})]})))

Duplicate Detection

(defprocessor :crm check-for-duplicates
  {:topics #{:crm/contact-created}}
  "Check for duplicates when contact is created."
  [context]
  (let [{:keys [contact-id]} (:event context)
        email (normalize-email (get-in (:event context) [:field-values :email]))
        existing (rm/get-contacts-by-email context email)]
    (if (seq (remove #(= contact-id (:id %)) existing))
      {:result/events
       [(->event
         {:type :crm/duplicate-detected
          :tags #{[:contact contact-id]}
          :body {:contact-id contact-id
                 :match-type :email
                 :match-value email}})]}
      {})))

Cascade Operations

(defprocessor :crm end-relationships-on-archive
  {:topics #{:crm/contact-archived}}
  "End all relationships when contact is archived."
  [context]
  (let [{:keys [contact-id]} (:event context)
        rels (rm/get-relationships-for-contact context contact-id)
        active-rels (filter #(nil? (:end-date %)) rels)]
    (if (seq active-rels)
      {:result/events
       (mapv (fn [rel]
               (->event
                {:type :crm/relationship-ended
                 :tags #{[:relationship (:id rel)]}
                 :body {:relationship-id (:id rel)
                        :end-date (str (java.time.LocalDate/now))
                        :reason "Contact archived"}}))
             active-rels)}
      {})))

Transform Source Data

(defprocessor :crm record-lead-attribution
  {:topics #{:crm/lead-captured}}
  "Map lead form type to attribution source."
  [context]
  (let [{:keys [contact-id form-type form-id]} (:event context)
        source (case form-type
                 :application :application_form
                 :intake :intake_form
                 :interest :interest_form
                 :manual_entry)]
    {:result/events
     [(->event
       {:type :crm/attribution-recorded
        :tags #{[:contact contact-id]}
        :body {:contact-id contact-id
               :attribution {:source source
                             :form-id form-id}}})]}))

Conventions

• Processors receive a context map with :event, :event-store, :tenant-id
• Use read models to check current state before acting
• Return {:result/events [...]} or {} (empty map)
• One processor can listen to multiple topics via the :topics set
• Multiple processors can listen to the same topic
• Normalize data before comparison (email lowercase, phone digits only)
• Include reason/context in generated events for audit trail

Reference Files

• components/user-service/src/ai/obney/workshop/user_service/core/todo_processors.clj - Email notifications
• components/crm-service/src/ai/obney/workshop/crm_service/core/todo_processors.clj - Attribution & duplicates