event-sourcing-coder

Rails 单体应用的事件溯源

记录重要的领域事件，并将其分派给专用处理器——这是一种务实的事件溯源方案，无需引入完整 CQRS/ES 基础设施的复杂性。

何时使用此技能

• 构建动态信息流或审计跟踪
• 将数据同步到外部服务（CRM、分析平台、Webhook）
• 自动执行由领域事件触发的工作流
• 将「发生了什么」与「该如何处理」解耦
• 跟踪用户操作，用于分析或调试

何时不使用事件

场景	更好的替代方案
单个模型上的简单回调	ActiveRecord 回调
仅有同步副作用	Service 对象或 ActiveInteraction
需要完整的事件重放／重建	专用事件溯源 Gem（Rails Event Store）
每种事件只有一个处理器	直接调用方法

核心概念

将记录与处理解耦：

User action → Record Event → Broadcast to Inboxes → Side Effects
                  ↓
              Queryable
              (audit trail)

设置

迁移

class CreateIssueEvents < ActiveRecord::Migration[8.0]
  def change
    create_table :issue_events do |t|
      t.references :issue, null: false, foreign_key: true
      t.references :actor, null: false, foreign_key: { to_table: :users }
      t.string :action, null: false
      t.jsonb :metadata, default: {}
      t.timestamps
    end

    add_index :issue_events, :action
    add_index :issue_events, :created_at
  end
end

事件模型

# app/models/issue/event.rb
module Issue::Event
  extend ActiveSupport::Concern

  ACTIONS = %w[
    created
    assigned
    status_changed
    commented
    closed
    reopened
  ].freeze

  included do
    belongs_to :issue
    belongs_to :actor, class_name: "User"

    validates :action, presence: true, inclusion: { in: ACTIONS }

    after_commit :broadcast_to_inboxes, on: :create
  end

  private

  def broadcast_to_inboxes
    Issue::Event::BroadcastJob.perform_later(self)
  end
end

class Issue::Event < ApplicationRecord
  include Issue::Event
end

记录事件

# app/models/issue.rb
class Issue < ApplicationRecord
  has_many :events, class_name: "Issue::Event", dependent: :destroy

  def record_event!(action:, actor:, metadata: {}, throttle: nil)
    return if throttle && recently_recorded?(action, throttle)

    events.create!(
      action: action,
      actor: actor,
      metadata: metadata
    )
  end

  private

  def recently_recorded?(action, duration)
    events.where(action: action)
          .where("created_at > ?", duration.ago)
          .exists?
  end
end

在应用中使用

# In a controller or interaction
issue.record_event!(
  action: "status_changed",
  actor: current_user,
  metadata: { from: "open", to: "in_progress" }
)

# With throttling (prevent duplicate events within timeframe)
issue.record_event!(
  action: "viewed",
  actor: current_user,
  throttle: 5.minutes
)

收件箱模式

收件箱是专用处理器，用于决定是否处理每种事件类型。

广播任务

# app/jobs/issue/event/broadcast_job.rb
class Issue::Event::BroadcastJob < ApplicationJob
  queue_as :events

  INBOXES = [
    Issue::Event::Inboxes::EmailNotifications,
    Issue::Event::Inboxes::SlackNotifications,
    Issue::Event::Inboxes::ExternalSync,
    Issue::Event::Inboxes::AutomationRules
  ].freeze

  def perform(event)
    INBOXES.each do |inbox_class|
      inbox_class.new(event).process
    end
  end
end

收件箱基类

# app/models/issue/event/inboxes/base.rb
module Issue::Event::Inboxes
  class Base
    attr_reader :event

    delegate :issue, :actor, :action, :metadata, to: :event

    def initialize(event)
      @event = event
    end

    def process
      return unless should_process?

      handle
    end

    private

    def should_process?
      raise NotImplementedError
    end

    def handle
      raise NotImplementedError
    end
  end
end

收件箱示例：电子邮件通知

# app/models/issue/event/inboxes/email_notifications.rb
module Issue::Event::Inboxes
  class EmailNotifications < Base
    NOTIFY_ACTIONS = %w[assigned commented status_changed].freeze

    private

    def should_process?
      action.in?(NOTIFY_ACTIONS) && recipients.any?
    end

    def handle
      recipients.each do |user|
        IssueMailer.event_notification(
          user: user,
          issue: issue,
          event: event
        ).deliver_later
      end
    end

    def recipients
      @recipients ||= issue.subscribers.where.not(id: actor.id)
    end
  end
end

收件箱示例：外部同步

# app/models/issue/event/inboxes/external_sync.rb
module Issue::Event::Inboxes
  class ExternalSync < Base
    SYNC_ACTIONS = %w[created status_changed closed].freeze

    private

    def should_process?
      action.in?(SYNC_ACTIONS) && issue.external_id.present?
    end

    def handle
      ExternalService::SyncIssueJob.perform_later(
        issue_id: issue.id,
        action: action,
        metadata: metadata
      )
    end
  end
end

动态信息流

事件可供查询，用于构建动态信息流：

# Recent activity on an issue
issue.events.includes(:actor).order(created_at: :desc).limit(20)

# User's recent activity across all issues
Issue::Event.where(actor: current_user)
            .includes(:issue)
            .order(created_at: :desc)
            .limit(50)

# Activity feed component
class Issue::ActivityFeedComponent < ViewComponent::Base
  def initialize(issue:)
    @events = issue.events.includes(:actor).order(created_at: :desc)
  end
end

与状态机集成

与 AASM 结合，实现自动事件记录：

class Issue < ApplicationRecord
  include AASM

  aasm column: :status do
    state :open, initial: true
    state :in_progress
    state :resolved
    state :closed

    event :start do
      transitions from: :open, to: :in_progress
      after { record_status_change("open", "in_progress") }
    end

    event :resolve do
      transitions from: :in_progress, to: :resolved
      after { record_status_change("in_progress", "resolved") }
    end
  end

  private

  def record_status_change(from, to)
    record_event!(
      action: "status_changed",
      actor: Current.user,
      metadata: { from: from, to: to }
    )
  end
end

测试

测试事件记录

RSpec.describe Issue do
  describe "#record_event!" do
    let(:issue) { create(:issue) }
    let(:user) { create(:user) }

    it "creates an event" do
      expect {
        issue.record_event!(action: "commented", actor: user)
      }.to change(issue.events, :count).by(1)
    end

    it "stores metadata" do
      issue.record_event!(
        action: "status_changed",
        actor: user,
        metadata: { from: "open", to: "closed" }
      )

      expect(issue.events.last.metadata).to eq(
        "from" => "open",
        "to" => "closed"
      )
    end

    context "with throttling" do
      it "prevents duplicate events within timeframe" do
        issue.record_event!(action: "viewed", actor: user)

        expect {
          issue.record_event!(action: "viewed", actor: user, throttle: 5.minutes)
        }.not_to change(issue.events, :count)
      end
    end
  end
end

测试收件箱

RSpec.describe Issue::Event::Inboxes::EmailNotifications do
  let(:event) { create(:issue_event, action: "assigned") }
  let(:inbox) { described_class.new(event) }

  describe "#process" do
    context "when issue has subscribers" do
      before { create(:subscription, issue: event.issue) }

      it "sends notification emails" do
        expect { inbox.process }
          .to have_enqueued_mail(IssueMailer, :event_notification)
      end
    end

    context "when action is not notifiable" do
      let(:event) { create(:issue_event, action: "viewed") }

      it "does not send emails" do
        expect { inbox.process }
          .not_to have_enqueued_mail(IssueMailer)
      end
    end
  end
end

最佳实践

保持事件不可变

# Good: Events are append-only facts
issue.record_event!(action: "status_changed", ...)

# Avoid: Never update or delete events
event.update!(action: "different")  # Don't do this

使用有意义的操作名称

# Good: Verb in past tense, describes what happened
ACTIONS = %w[created assigned commented resolved closed reopened]

# Avoid: Vague or present-tense actions
ACTIONS = %w[update change action do_thing]

在元数据中存储上下文

# Good: Capture context at event time
record_event!(
  action: "status_changed",
  actor: current_user,
  metadata: {
    from: previous_status,
    to: new_status,
    reason: params[:reason],
    triggered_by: "manual"  # vs "automation"
  }
)

# Avoid: Relying on current state (it changes)
record_event!(action: "status_changed", actor: current_user)
# Later: "What was the previous status?" - Unknown!

异步处理事件

# Good: Inboxes process in background jobs
after_commit :broadcast_to_inboxes, on: :create

def broadcast_to_inboxes
  BroadcastJob.perform_later(self)
end

# Avoid: Synchronous processing blocks the request
after_create :process_all_inboxes  # Slow!

详细参考

高级模式：

• references/event-model.md - 多态事件、自定义类型、关注点
• references/inbox-pattern.md - 收件箱组合、错误处理、重试
• references/use-cases.md - 动态信息流、Webhook、自动化规则