mybook

Kata: 通知基盤 — マルチチャネル配信

課題の提示

「タクミさん、最後にユーザーに何かを"伝えた"のはいつ？」

ナオミの質問の意味がわからず、タクミは考えた。コードで？それとも人として？

「システムの話よ。予約確認メール、在庫切れのプッシュ通知、配送SMSのアラート。これを設計するの」

Kata 5: マルチチャネル通知基盤

複数のサービスから通知を送る基盤を作りたい。

  • 通知チャネル: Email / プッシュ通知（iOS/Android）/ SMS
  • 送信数: 月間500万通
  • 各ユーザーがチャネル・タイミングを設定できる
  • 送信失敗時はリトライする
  • 送信ログを保持する（法令対応）
  • 通知の重複送信を防ぐ
  • サービスが増えても通知基盤を変更不要にする

タクミは最後の要件を読み返した。「サービスが増えても通知基盤を変更不要」。これは難しい。

「これは、疎結合の問題ですね」

「そう。予約サービスが増えたとき、ECサービスが増えたとき、通知基盤のコードを変更せずに対応できるか。その設計が今日の核心よ」

設計判断

結合方式の比較

Loading diagram...

直接呼び出しの問題: 予約サービスが通知サービスのAPIを知っている。通知サービスが落ちたら予約も失敗する可能性がある。

判断: イベント駆動アーキテクチャを採用する。サービスはイベントを発火するだけ。通知基盤はそれを受け取る。

AWSサービスの役割分担

Loading diagram...
AWSサービス役割
SNSイベントの発行・ファンアウト
SQS非同期キュー、リトライ管理
SES大量メール配信
Lambda軽量な通知処理（オプション）

実装

通知イベントの定義

# app/events/notification_event.rb
# 全通知イベントの基底クラス
class NotificationEvent
  attr_reader :user_id, :data, :occurred_at
 
  def initialize(user_id:, data: {})
    @user_id = user_id
    @data = data
    @occurred_at = Time.current
  end
 
  def event_type
    self.class.name.underscore
  end
 
  def to_sns_message
    {
      event_type: event_type,
      user_id: user_id,
      data: data,
      occurred_at: occurred_at.iso8601
    }.to_json
  end
end
 
# 予約確認イベント
class ReservationConfirmedEvent < NotificationEvent
  def initialize(user_id:, reservation_id:, hotel_name:, check_in:, check_out:)
    super(
      user_id: user_id,
      data: {
        reservation_id: reservation_id,
        hotel_name: hotel_name,
        check_in: check_in,
        check_out: check_out
      }
    )
  end
end
 
# 在庫切れアラートイベント
class ProductOutOfStockEvent < NotificationEvent
  def initialize(user_id:, product_id:, product_name:)
    super(
      user_id: user_id,
      data: { product_id: product_id, product_name: product_name }
    )
  end
end

SNS へのイベント発行

# app/services/event_publisher.rb
class EventPublisher
  SNS_TOPIC_ARN = ENV["NOTIFICATION_SNS_TOPIC_ARN"]
 
  def self.publish(event)
    sns_client.publish(
      topic_arn: SNS_TOPIC_ARN,
      message: event.to_sns_message,
      message_attributes: {
        "event_type" => {
          data_type: "String",
          string_value: event.event_type
        }
      }
    )
  rescue Aws::SNS::Errors::ServiceError => e
    # SNS発行失敗はログして継続（通知は重要だが、ビジネス処理を止めない）
    Rails.logger.error("SNS publish failed: #{e.message}, event: #{event.event_type}")
    ErrorTracker.capture(e)
  end
 
  private
 
  def self.sns_client
    @sns_client ||= Aws::SNS::Client.new(region: ENV["AWS_REGION"])
  end
end
 
# 予約サービスからの使用例
class ReservationsController < ApplicationController
  def create
    reservation = room.reserve_with_lock!(current_user, check_in, check_out)
 
    # イベントを発行（通知は非同期で処理される）
    EventPublisher.publish(
      ReservationConfirmedEvent.new(
        user_id: current_user.id,
        reservation_id: reservation.id,
        hotel_name: room.hotel.name,
        check_in: reservation.check_in,
        check_out: reservation.check_out
      )
    )
 
    redirect_to reservation, notice: "予約が完了しました"
  end
end

SQS からのメッセージ処理

# app/jobs/notification_processor_job.rb
class NotificationProcessorJob < ApplicationJob
  queue_as :notifications
 
  def perform(sqs_message)
    event = JSON.parse(sqs_message["body"])
    user = User.find(event["user_id"])
 
    # ユーザーの通知設定を確認
    preferences = user.notification_preferences
 
    # 重複送信チェック
    idempotency_key = "notification:#{event['event_type']}:#{event['user_id']}:#{sqs_message['messageId']}"
    return if $redis.exists(idempotency_key)
 
    # チャネルごとに送信
    send_email(user, event) if preferences.email_enabled?
    send_push(user, event) if preferences.push_enabled?
    send_sms(user, event) if preferences.sms_enabled? && event_requires_sms?(event)
 
    # 送信ログを保存
    NotificationLog.create!(
      user_id: user.id,
      event_type: event["event_type"],
      channels: preferences.enabled_channels,
      sent_at: Time.current,
      message_id: sqs_message["messageId"]
    )
 
    # 重複防止フラグ（24時間）
    $redis.setex(idempotency_key, 24.hours, 1)
  end
 
  private
 
  def send_email(user, event)
    template = NotificationTemplate.find_by(event_type: event["event_type"], channel: :email)
    return unless template
 
    NotificationMailer
      .with(user: user, template: template, data: event["data"])
      .notify
      .deliver_later
  end
 
  def send_push(user, event)
    return unless user.device_tokens.any?
 
    template = NotificationTemplate.find_by(event_type: event["event_type"], channel: :push)
    return unless template
 
    user.device_tokens.each do |token|
      PushNotificationService.send(
        token: token.value,
        platform: token.platform,
        title: template.render_title(event["data"]),
        body: template.render_body(event["data"])
      )
    end
  end
 
  def event_requires_sms?(event)
    %w[reservation_confirmed order_shipped].include?(event["event_type"])
  end
end

ユーザーの通知設定

# app/models/notification_preference.rb
class NotificationPreference < ApplicationRecord
  belongs_to :user
 
  # 通知設定をJSON型で保存
  store_accessor :settings, :email_enabled, :push_enabled, :sms_enabled,
                            :quiet_hours_start, :quiet_hours_end,
                            :digest_mode  # まとめて送るモード
 
  def email_enabled?
    settings["email_enabled"] != false
  end
 
  def push_enabled?
    settings["push_enabled"] != false
  end
 
  def sms_enabled?
    settings["sms_enabled"] == true
  end
 
  def enabled_channels
    channels = []
    channels << "email" if email_enabled?
    channels << "push" if push_enabled?
    channels << "sms" if sms_enabled?
    channels
  end
 
  def in_quiet_hours?
    return false unless quiet_hours_start && quiet_hours_end
    current_hour = Time.current.in_time_zone(user.timezone).hour
    (quiet_hours_start..quiet_hours_end).include?(current_hour)
  end
end

SES によるメール送信

# app/mailers/notification_mailer.rb
class NotificationMailer < ApplicationMailer
  def notify
    @user = params[:user]
    @template = params[:template]
    @data = params[:data]
 
    mail(
      to: @user.email,
      subject: @template.render_subject(@data),
      from: "notifications@example.com"
    ) do |format|
      format.html { render html: @template.render_html(@data).html_safe }
      format.text { render plain: @template.render_text(@data) }
    end
  end
end
 
# config/environments/production.rb
config.action_mailer.delivery_method = :ses
config.action_mailer.ses_settings = {
  region: ENV["AWS_REGION"],
  access_key_id: ENV["AWS_ACCESS_KEY_ID"],
  secret_access_key: ENV["AWS_SECRET_ACCESS_KEY"]
}

INFO

SES は月間62,000通まで無料（EC2から送信の場合）。月間500万通なら費用が発生するが、1,000通あたり$0.10程度と非常に安価。バウンスとクレームの管理をSESのダッシュボードで行う。

AWSインフラ構成（詳細）

Loading diagram...

SQS のデッドレターキュー設定

{
  "QueueName": "notification-email-queue",
  "Attributes": {
    "VisibilityTimeout": "60",
    "MessageRetentionPeriod": "86400",
    "ReceiveMessageWaitTimeSeconds": "20",
    "RedrivePolicy": {
      "deadLetterTargetArn": "arn:aws:sqs:...:notification-email-dlq",
      "maxReceiveCount": "3"
    }
  }
}

WARNING

デッドレターキュー（DLQ）は必須設定。3回失敗したメッセージは DLQ に移動し、CloudWatch アラームで検知する。DLQ のメッセージを確認してバグを修正した後、再処理する仕組みを必ず用意する。

送信ログと法令対応

# app/models/notification_log.rb
class NotificationLog < ApplicationRecord
  belongs_to :user
  belongs_to :notification_template, optional: true
 
  validates :event_type, presence: true
  validates :sent_at, presence: true
 
  # 7年間保持（法令要件）
  scope :recent, -> { where("sent_at > ?", 7.years.ago) }
 
  # S3へのアーカイブ（月次バッチ）
  def self.archive_old_logs
    old_logs = where("sent_at < ?", 1.year.ago)
    # S3にJSONLファイルとしてエクスポート
    ArchiveToS3Job.perform_later(
      records: old_logs.as_json,
      key: "notification-logs/#{Date.current.strftime('%Y/%m')}/archive.jsonl"
    )
    old_logs.delete_all
  end
end

振り返り

「500万通の送信、全部成功する？」とナオミが聞いた。

「しません。必ず一定数は失敗します」

「そう。だから設計に"失敗"を組み込む。SQS のリトライ、DLQ、ログの保持。通知は失敗しても良い。でも失敗を見えるようにすることが設計の責務よ」

INFO

Kata 5 の学び: 通知は非同期でいい。でも「送った」「失敗した」「再送した」を追跡できなければ、システムは信頼されない。非機能要件（ログ・リトライ・DLQ）が設計の半分を占める。

トレードオフの記録

決定メリットデメリット
SNS → SQS疎結合、スケーラブルインフラ複雑さ増加
チャネル別SQS独立スケール、障害分離キュー数が増える
Redis 重複防止冪等性保証Redisに依存

「次は、そのメールに添付したいファイルをどう保存するか。ファイルアップロードのKata」

前の章Kata: 予約システム — 排他制御と整合性
次の章Kata: ファイルアップロード — 大容量データの扱い