Add Pipelines and Deals read models to CRM app

Author: andrzejkrzywda

apps/crm/.mutant.yml

@@ -11,8 +11,17 @@ matcher:
   subjects:
     - Companies*
     - Contacts*
+    - Pipelines*
+    - Deals*
   ignore:
     - Companies::Company
     - Companies::Configuration#call
     - Contacts::Contact
     - Contacts::Configuration#call
+    - Pipelines::Pipeline
+    - Pipelines::Stage
+    - Pipelines::Configuration#call
+    - Pipelines.stages_for
+    - Deals::Deal
+    - Deals::Configuration#call
+    - Deals.for_pipeline

apps/crm/app/read_models/deals/configuration.rb

@@ -0,0 +1,55 @@
+module Deals
+  class Deal < ApplicationRecord
+    self.table_name = "deals"
+  end
+  private_constant :Deal
+
+  def self.all
+    Deal.order(id: :asc)
+  end
+
+  def self.find_by_uid(uid)
+    Deal.find_by!(uid: uid)
+  end
+
+  def self.for_pipeline(pipeline_uid)
+    Deal.where(pipeline_uid: pipeline_uid).order(id: :asc)
+  end
+
+  class CreateDeal
+    def call(event)
+      Deal.create!(
+        uid: event.data.fetch(:deal_id),
+        pipeline_uid: event.data.fetch(:pipeline_id),
+        name: event.data.fetch(:name)
+      )
+    end
+  end
+
+  class SetValue
+    def call(event)
+      Deal.find_by!(uid: event.data.fetch(:deal_id)).update!(value: event.data.fetch(:value))
+    end
+  end
+
+  class SetExpectedCloseDate
+    def call(event)
+      Deal.find_by!(uid: event.data.fetch(:deal_id)).update!(expected_close_date: event.data.fetch(:expected_close_date))
+    end
+  end
+
+  class MoveToStage
+    def call(event)
+      Deal.find_by!(uid: event.data.fetch(:deal_id)).update!(stage: event.data.fetch(:stage))
+    end
+  end
+
+  class Configuration
+    def call(event_store)
+      event_store.subscribe(CreateDeal.new, to: [Crm::DealCreated])
+      event_store.subscribe(SetValue.new, to: [Crm::DealValueSet])
+      event_store.subscribe(SetExpectedCloseDate.new, to: [Crm::DealExpectedCloseDateSet])
+      event_store.subscribe(MoveToStage.new, to: [Crm::DealMovedToStage])
+    end
+  end
+end

apps/crm/app/read_models/pipelines/configuration.rb

@@ -0,0 +1,58 @@
+module Pipelines
+  class Pipeline < ApplicationRecord
+    self.table_name = "pipelines"
+  end
+  private_constant :Pipeline
+
+  class Stage < ApplicationRecord
+    self.table_name = "pipeline_stages"
+  end
+  private_constant :Stage
+
+  def self.all
+    Pipeline.order(id: :asc)
+  end
+
+  def self.find_by_uid(uid)
+    Pipeline.find_by!(uid: uid)
+  end
+
+  def self.stages_for(pipeline_uid)
+    Stage.where(pipeline_uid: pipeline_uid).order(id: :asc)
+  end
+
+  class CreatePipeline
+    def call(event)
+      Pipeline.create!(
+        uid: event.data.fetch(:pipeline_id),
+        name: event.data.fetch(:name)
+      )
+    end
+  end
+
+  class AddStage
+    def call(event)
+      Stage.create!(
+        pipeline_uid: event.data.fetch(:pipeline_id),
+        stage_name: event.data.fetch(:stage_name)
+      )
+    end
+  end
+
+  class RemoveStage
+    def call(event)
+      Stage.find_by!(
+        pipeline_uid: event.data.fetch(:pipeline_id),
+        stage_name: event.data.fetch(:stage_name)
+      ).destroy!
+    end
+  end
+
+  class Configuration
+    def call(event_store)
+      event_store.subscribe(CreatePipeline.new, to: [Crm::PipelineCreated])
+      event_store.subscribe(AddStage.new, to: [Crm::StageAddedToPipeline])
+      event_store.subscribe(RemoveStage.new, to: [Crm::StageRemovedFromPipeline])
+    end
+  end
+end

apps/crm/db/migrate/20260217112240_create_pipelines.rb

@@ -0,0 +1,11 @@
+class CreatePipelines < ActiveRecord::Migration[8.0]
+  def change
+    create_table :pipelines do |t|
+      t.uuid :uid, null: false
+      t.string :name, null: false
+
+      t.timestamps
+    end
+    add_index :pipelines, :uid, unique: true
+  end
+end

apps/crm/db/migrate/20260217112241_create_pipeline_stages.rb

@@ -0,0 +1,11 @@
+class CreatePipelineStages < ActiveRecord::Migration[8.0]
+  def change
+    create_table :pipeline_stages do |t|
+      t.uuid :pipeline_uid, null: false
+      t.string :stage_name, null: false
+
+      t.timestamps
+    end
+    add_index :pipeline_stages, [:pipeline_uid, :stage_name], unique: true
+  end
+end

apps/crm/db/migrate/20260217112242_create_deals.rb

@@ -0,0 +1,15 @@
+class CreateDeals < ActiveRecord::Migration[8.0]
+  def change
+    create_table :deals do |t|
+      t.uuid :uid, null: false
+      t.uuid :pipeline_uid, null: false
+      t.string :name, null: false
+      t.integer :value
+      t.string :expected_close_date
+      t.string :stage
+
+      t.timestamps
+    end
+    add_index :deals, :uid, unique: true
+  end
+end

apps/crm/db/schema.rb

@@ -6,6 +6,8 @@ def call(event_store, command_bus)
     enable_res_infra_event_linking(event_store)
     enable_contacts_read_model(event_store)
     enable_companies_read_model(event_store)
+    enable_pipelines_read_model(event_store)
+    enable_deals_read_model(event_store)
 
     Crm::Configuration.new.call(event_store, command_bus)
   end
@@ -20,6 +22,14 @@ def enable_contacts_read_model(event_store)
     Contacts::Configuration.new.call(event_store)
   end
 
+  def enable_pipelines_read_model(event_store)
+    Pipelines::Configuration.new.call(event_store)
+  end
+
+  def enable_deals_read_model(event_store)
+    Deals::Configuration.new.call(event_store)
+  end
+
   def enable_res_infra_event_linking(event_store)
     [
       RailsEventStore::LinkByEventType.new,

apps/crm/lib/configuration.rb

@@ -0,0 +1,90 @@
+require "test_helper"
+
+module Deals
+  class DealsTest < InMemoryRESTestCase
+    cover "Deals*"
+
+    def test_deal_created
+      create_deal(deal_id, pipeline_id, "Big Deal")
+
+      assert_equal(1, Deals.all.count)
+      deal = Deals.find_by_uid(deal_id)
+      assert_equal("Big Deal", deal.name)
+      assert_equal(pipeline_id, deal.pipeline_uid)
+    end
+
+    def test_multiple_deals
+      create_deal(deal_id, pipeline_id, "Big Deal")
+      create_deal(other_deal_id, pipeline_id, "Small Deal")
+
+      assert_equal(2, Deals.all.count)
+      assert_equal(["Big Deal", "Small Deal"], Deals.all.map(&:name))
+    end
+
+    def test_for_pipeline
+      other_pipeline_id = SecureRandom.uuid
+      create_deal(deal_id, pipeline_id, "Big Deal")
+      create_deal(other_deal_id, other_pipeline_id, "Small Deal")
+
+      assert_equal(["Big Deal"], Deals.for_pipeline(pipeline_id).map(&:name))
+      assert_equal(["Small Deal"], Deals.for_pipeline(other_pipeline_id).map(&:name))
+    end
+
+    def test_value_set
+      create_deal(deal_id, pipeline_id, "Big Deal")
+      create_deal(other_deal_id, pipeline_id, "Small Deal")
+      set_value(deal_id, 10_000)
+
+      assert_equal(10_000, Deals.find_by_uid(deal_id).value)
+      assert_nil(Deals.find_by_uid(other_deal_id).value)
+    end
+
+    def test_expected_close_date_set
+      create_deal(deal_id, pipeline_id, "Big Deal")
+      create_deal(other_deal_id, pipeline_id, "Small Deal")
+      set_expected_close_date(deal_id, "2026-03-01")
+
+      assert_equal("2026-03-01", Deals.find_by_uid(deal_id).expected_close_date)
+      assert_nil(Deals.find_by_uid(other_deal_id).expected_close_date)
+    end
+
+    def test_moved_to_stage
+      create_deal(deal_id, pipeline_id, "Big Deal")
+      create_deal(other_deal_id, pipeline_id, "Small Deal")
+      move_to_stage(deal_id, "Negotiation")
+
+      assert_equal("Negotiation", Deals.find_by_uid(deal_id).stage)
+      assert_nil(Deals.find_by_uid(other_deal_id).stage)
+    end
+
+    private
+
+    def deal_id
+      @deal_id ||= SecureRandom.uuid
+    end
+
+    def other_deal_id
+      @other_deal_id ||= SecureRandom.uuid
+    end
+
+    def pipeline_id
+      @pipeline_id ||= SecureRandom.uuid
+    end
+
+    def create_deal(uid, pipeline_uid, name)
+      event_store.publish(Crm::DealCreated.new(data: { deal_id: uid, pipeline_id: pipeline_uid, name: name }))
+    end
+
+    def set_value(uid, value)
+      event_store.publish(Crm::DealValueSet.new(data: { deal_id: uid, value: value }))
+    end
+
+    def set_expected_close_date(uid, date)
+      event_store.publish(Crm::DealExpectedCloseDateSet.new(data: { deal_id: uid, expected_close_date: date }))
+    end
+
+    def move_to_stage(uid, stage)
+      event_store.publish(Crm::DealMovedToStage.new(data: { deal_id: uid, stage: stage }))
+    end
+  end
+end