CQRS — コマンドとクエリの分離

読み書きは本質的に非対称だ

プロジェクト開始から2ヶ月。カオリはパフォーマンスの問題に直面していた。

「管理画面の注文一覧が遅い。売上ダッシュボードも重い。でも注文作成自体は速い」

ログを見ると明らかだった。注文一覧のクエリは、注文・ユーザー・商品・在庫・クーポンを複数テーブルでJOINし、集計まで行っている。一方、注文作成は単純な書き込みだ。

これが読み書きの非対称性だ。

書き込み（コマンド）の特性:

整合性が重要（トランザクション）
検証ロジックが複雑
スループットより正確性

読み取り（クエリ）の特性:

高速なレスポンスが必要
様々な形状のデータが必要（詳細、一覧、ダッシュボード）
スケールが必要（読み取りの方が圧倒的に多い）

INFO

CQRS（Command Query Responsibility Segregation）は、データの「変更」（コマンド）と「読み取り」（クエリ）を明確に分離する設計パターンです。Bertrand Meyerの「コマンドクエリ分離原則（CQS）」をシステムレベルに拡張したものです。

CQSとCQRSの違い

# CQS: メソッドレベルの分離
class OrderService
  # コマンド: 状態を変更し、値を返さない
  def create_order(user:, product:, quantity:)
    Order.create!(user: user, product: product, quantity: quantity)
    nil  # 戻り値なし（または成功/失敗のみ）
  end
  
  # クエリ: 状態を変更せず、値を返す
  def find_orders_for_user(user_id)
    Order.where(user_id: user_id).order(created_at: :desc)
  end
end
 
# CQRS: システムアーキテクチャレベルの分離
# コマンドとクエリが別々のモデル・DB・サービスを持つ

Railsでの基本的なCQRS実装

コマンド側（書き込みモデル）

# app/commands/place_order_command.rb
class PlaceOrderCommand
  attr_reader :user_id, :product_id, :quantity, :coupon_code
  
  def initialize(user_id:, product_id:, quantity:, coupon_code: nil)
    @user_id = user_id
    @product_id = product_id
    @quantity = Integer(quantity)
    @coupon_code = coupon_code
    
    validate!
  end
  
  private
  
  def validate!
    raise ArgumentError, "数量は1以上" unless quantity.positive?
    raise ArgumentError, "数量は100以下" if quantity > 100
  end
end
 
# app/command_handlers/place_order_handler.rb
class PlaceOrderHandler
  def initialize(order_repository:, product_repository:, event_store:)
    @order_repository = order_repository
    @product_repository = product_repository
    @event_store = event_store
  end
  
  def handle(command)
    product = @product_repository.find(command.product_id)
    
    raise InsufficientInventoryError unless product.stock_count >= command.quantity
    
    # 書き込みモデルは正規化された形式で保存
    order = Order.new(
      user_id: command.user_id,
      product_id: command.product_id,
      quantity: command.quantity,
      unit_price: product.price,
      status: 'pending'
    )
    
    @order_repository.save(order)
    
    # ドメインイベントを発行（クエリモデルの更新をトリガー）
    @event_store.publish(OrderPlaced.new(
      order_id: order.id,
      user_id: order.user_id,
      product_id: order.product_id,
      quantity: order.quantity,
      total_price: order.quantity * product.price,
      occurred_at: Time.current
    ))
    
    order
  end
end

クエリ側（読み取りモデル）

クエリモデルは非正規化されており、読み取りに最適化されている。

# app/query_models/order_list_view.rb
# これはReadOnlyのモデル（write しない）
class OrderListView < ApplicationRecord
  self.table_name = 'order_list_views'
  
  # 非正規化されたビュー（JOINなしで全情報が取得できる）
  # id, user_id, user_name, user_email,
  # product_id, product_name, product_image_url,
  # quantity, unit_price, total_price,
  # status, status_label_ja,
  # created_at, updated_at_at_formatted
 
  scope :for_user, ->(user_id) { where(user_id: user_id) }
  scope :recent, -> { order(created_at: :desc) }
  scope :by_status, ->(status) { where(status: status) }
  
  def self.search(keyword)
    where('product_name ILIKE ? OR user_name ILIKE ?', "%#{keyword}%", "%#{keyword}%")
  end
end
 
# app/query_models/sales_dashboard_view.rb
class SalesDashboardView < ApplicationRecord
  self.table_name = 'sales_dashboard_views'
  
  # 集計済みのビュー
  # date, total_orders, total_revenue, avg_order_value,
  # top_product_id, top_product_name
 
  scope :this_month, -> { where(date: Date.current.beginning_of_month..) }
  scope :last_30_days, -> { where(date: 30.days.ago..) }
end

クエリオブジェクト

# app/queries/order_list_query.rb
class OrderListQuery
  def initialize(relation = OrderListView.all)
    @relation = relation
  end
  
  def for_user(user_id)
    self.class.new(@relation.for_user(user_id))
  end
  
  def with_status(status)
    self.class.new(@relation.by_status(status))
  end
  
  def search(keyword)
    self.class.new(@relation.search(keyword))
  end
  
  def paginate(page:, per: 20)
    @relation.recent.offset((page - 1) * per).limit(per)
  end
end
 
# 使い方
orders = OrderListQuery.new
  .for_user(current_user.id)
  .with_status('pending')
  .paginate(page: 1)

クエリモデルの更新

コマンド側の変更をクエリモデルに反映させる「プロジェクター」。

# app/projectors/order_list_projector.rb
class OrderListProjector
  def on_order_placed(event)
    user = User.find(event.user_id)
    product = ProductRecord.find(event.product_id)
    
    OrderListView.create!(
      id: event.order_id,
      user_id: event.user_id,
      user_name: user.full_name,
      user_email: user.email,
      product_id: event.product_id,
      product_name: product.name,
      product_image_url: product.image_url,
      quantity: event.quantity,
      unit_price: product.price,
      total_price: event.total_price,
      status: 'pending',
      status_label_ja: '注文受付',
      created_at: event.occurred_at
    )
  end
  
  def on_order_confirmed(event)
    OrderListView.find(event.order_id).update!(
      status: 'confirmed',
      status_label_ja: '確認済み'
    )
  end
  
  def on_order_shipped(event)
    OrderListView.find(event.order_id).update!(
      status: 'shipped',
      status_label_ja: '発送済み',
      shipped_at: event.occurred_at,
      tracking_number: event.tracking_number
    )
  end
end

AWSでのCQRS構成

Loading diagram...

# AWSリソース構成（Terraform風）
resources:
  # コマンド側
  command_api:
    type: ECS Fargate
    env: RAILS_ENV=production
    min_count: 2
    max_count: 10
    
  write_db:
    type: RDS PostgreSQL
    instance_class: db.r6g.large  # 書き込み最適化
    multi_az: true
    
  # イベント配信
  event_queue:
    type: SQS
    visibility_timeout: 300
    
  # クエリ側  
  query_api:
    type: ECS Fargate  
    min_count: 3       # 読み取りは多いのでスケール大
    max_count: 50
    
  read_cache:
    type: ElastiCache Redis
    node_type: cache.r6g.large
    
  projector:
    type: Lambda
    trigger: SQS
    batch_size: 10

結果整合性

CQRSの重要な特性は「結果整合性」だ。

# コマンドが実行される
PlaceOrderHandler.new(...).handle(place_order_command)
# → RDS に書き込み
# → SQS にイベント発行
# → Lambda が数百ms後にクエリモデルを更新
 
# この数百ms間、クエリモデルは古いデータを返す
# これが「結果整合性」

WARNING

CQRSを導入すると、書き込みと読み取りの間に時間的なギャップ（数百ms〜数秒）が生じます。「注文したのにすぐ一覧に出ない」という体験を許容できるユースケースかどうか、設計前に確認が必要です。

ユーザー体験での対策

# app/controllers/api/v1/orders_controller.rb
class Api::V1::OrdersController < ApplicationController
  def create
    result = PlaceOrderHandler.new(...).handle(command)
    
    if result.success?
      # 注文IDをキャッシュに保存（ポーリング用）
      cache_key = "pending_order_#{result.order_id}"
      Rails.cache.write(cache_key, result.order.to_h, expires_in: 30.seconds)
      
      render json: {
        order_id: result.order_id,
        status: 'processing',
        # フロントエンドに「まだ反映中」を伝える
        message: '注文を受け付けました。一覧への反映まで少々お待ちください。'
      }, status: :accepted  # 202 Accepted
    end
  end
end

シンプルなCQRS（同一DB）

「フルのCQRSは複雑すぎる」という場合、同じDBでもモデルを分けることができる。

# 同一DBでCQRSの思想を適用（シンプル版）
 
# 書き込みモデル：ビジネスルールを持つ
class Order < ApplicationRecord
  validates :quantity, numericality: { greater_than: 0 }
  
  def confirm!
    raise "確認不可" unless pending?
    update!(status: 'confirmed')
    publish_event(:order_confirmed)
  end
end
 
# 読み取りモデル：クエリに最適化されたメソッドのみ
class OrderQuery < ApplicationRecord
  self.table_name = 'orders'
  
  # 書き込みメソッドを封印
  def readonly? = true
  
  # 読み取り専用の便利メソッド群
  scope :with_details, -> { includes(:user, :product) }
  scope :for_dashboard, -> { select('DATE(created_at) as date, COUNT(*) as count, SUM(total_price) as revenue').group('DATE(created_at)') }
  
  def self.search_full_text(query)
    joins(:product).where('products.name ILIKE ?', "%#{query}%")
  end
end

どこまでCQRSを導入するか

カオリはチームに判断基準を示した。

レベル1: メソッドレベルのCQS（最小コスト）
  → コマンドとクエリのメソッドを分ける
  → 全プロジェクトで推奨

レベル2: モデルレベルのCQRS（中程度のコスト）
  → 書き込みモデルと読み取りモデルを別クラスにする
  → 読み書きの性質が大きく異なるエンティティに

レベル3: ストアレベルのCQRS（高コスト）
  → 書き込みDBと読み取りDB/キャッシュを分離
  → 高トラフィック・複雑なクエリが必要な場合

レベル4: フルCQRS + イベントソーシング（最高コスト）
  → 次章のイベントソーシングと組み合わせ
  → 監査ログ・タイムトラベルが必要な場合

「今の私たちは、レベル2から始める。管理画面の重いクエリだけ読み取りモデルに切り出す」

次章では、CQRSとペアで使われる「イベントソーシング」を学ぶ。状態ではなく「事実（イベント）」を記録することで、タイムトラベルと完全な監査ログを実現する設計を見ていこう。