設計問題: 通知システム — Push/Email/SMSの統合配信

# app/services/notification_service.rb
class NotificationService
  PRIORITY_TOPICS = {
    critical: 'notifications-critical',
    high:     'notifications-high',
    normal:   'notifications-normal',
    low:      'notifications-low'
  }.freeze
 
  def self.send(user_id:, type:, priority: :normal, channels:, data: {})
    user = User.find(user_id)
 
    # 1. オプトアウト・通知設定チェック
    settings = UserNotificationSetting.find_or_initialize_by(
      user_id: user_id,
      notification_type: type
    )
    return if settings.globally_opted_out?
 
    # 2. レート制限チェック（同一ユーザーへの過剰送信を防ぐ）
    return unless RateLimiter.allow?(user_id: user_id, priority: priority)
 
    # 3. 有効なチャネルにフィルタリング
    valid_channels = channels.select do |ch|
      settings.channel_enabled?(ch) && user.token_exists_for?(ch)
    end
    return if valid_channels.empty?
 
    # 4. 冪等性キーを生成（重複送信防止）
    idempotency_key = IdempotencyKeyGenerator.call(
      user_id: user_id, type: type, data: data
    )
    return unless DeduplicationService.check_and_mark(idempotency_key)
 
    # 5. 通知ログを作成
    notification = Notification.create!(
      user_id:           user_id,
      notification_type: type,
      idempotency_key:   idempotency_key,
      status:            'pending',
      priority:          priority,
      payload:           data
    )
 
    # 6. 優先度別キューへ publish
    valid_channels.each do |channel|
      SqsPublisher.publish(
        queue_url: PRIORITY_TOPICS[priority],
        message: {
          notification_id: notification.id,
          channel:         channel,
          user_id:         user_id
        }
      )
    end
  end
end

# config/workers/notification_worker_config.rb
WORKER_CONFIG = {
  critical: { concurrency: 20, visibility_timeout: 30 },
  high:     { concurrency: 10, visibility_timeout: 60 },
  normal:   { concurrency: 5,  visibility_timeout: 120 },
  low:      { concurrency: 2,  visibility_timeout: 300 }
}.freeze
 
# 優先度の判定ロジック
module NotificationPriority
  CRITICAL_TYPES = %i[
    payment_failed account_suspended security_alert
    password_changed two_factor_disabled
  ].freeze
 
  HIGH_TYPES = %i[
    new_message mention direct_reply order_shipped
  ].freeze
 
  def self.resolve(type)
    return :critical if CRITICAL_TYPES.include?(type.to_sym)
    return :high     if HIGH_TYPES.include?(type.to_sym)
    :normal
  end
end

# app/models/device_token.rb
class DeviceToken < ApplicationRecord
  belongs_to :user
  enum platform: { ios: 0, android: 1 }
  enum status: { active: 0, expired: 1, invalid: 2 }
 
  # ユーザーが複数デバイスを持つことを前提に設計
  scope :active_for_user, ->(user_id) {
    where(user_id: user_id, status: :active)
  }
 
  # トークン登録（同一デバイスの重複登録を防ぐ）
  def self.register(user_id:, token:, platform:)
    find_or_initialize_by(token: token).tap do |dt|
      dt.update!(
        user_id:      user_id,
        platform:     platform,
        status:       :active,
        last_seen_at: Time.current
      )
    end
  end
 
  # APNs/FCM からの失敗フィードバックを処理
  def self.handle_delivery_failure(token:, error_code:)
    record = find_by(token: token)
    return unless record
 
    case error_code
    when 'BadDeviceToken', 'Unregistered', 'NotRegistered'
      record.update!(status: :invalid, invalidated_at: Time.current)
      Rails.logger.warn("Device token invalidated: #{token[0..8]}...")
    when 'DeviceTokenNotForTopic'
      # 別アプリのトークンが混入している（バグの可能性）
      record.update!(status: :invalid)
      Sentry.capture_message("Wrong topic for token: #{token[0..8]}")
    end
  end
end

# app/workers/push_notification_worker.rb
class PushNotificationWorker
  include Sidekiq::Worker
  sidekiq_options retry: 3, queue: 'push'
 
  def perform(notification_id)
    notification = Notification.find(notification_id)
    tokens = DeviceToken.active_for_user(notification.user_id)
 
    return notification.update!(status: 'no_token') if tokens.empty?
 
    results = tokens.map { |token| deliver_to_token(notification, token) }
 
    if results.any? { |r| r[:success] }
      notification.update!(status: 'delivered', delivered_at: Time.current)
    else
      notification.update!(status: 'failed')
      raise "All tokens failed for notification #{notification_id}"
    end
  end
 
  private
 
  def deliver_to_token(notification, device_token)
    result = case device_token.platform
             when 'ios'
               ApnsClient.send(
                 token: device_token.token,
                 title: notification.title,
                 body:  notification.body,
                 badge: notification.user.unread_notifications_count,
                 data:  notification.payload
               )
             when 'android'
               FcmClient.send(
                 token:        device_token.token,
                 title:        notification.title,
                 body:         notification.body,
                 data:         notification.payload,
                 click_action: 'FLUTTER_NOTIFICATION_CLICK'
               )
             end
 
    unless result[:success]
      DeviceToken.handle_delivery_failure(
        token:      device_token.token,
        error_code: result[:error_code]
      )
    end
 
    result
  end
end

# app/models/notification_template.rb
class NotificationTemplate < ApplicationRecord
  # DB設計: notification_templates
  #   id, notification_type, locale,
  #   title_template, body_template,
  #   email_subject_template, email_html_template,
  #   created_at, updated_at
 
  def self.render(type:, locale:, variables: {})
    template = find_by!(notification_type: type, locale: locale)
 
    {
      title: Liquid::Template.parse(template.title_template).render(variables),
      body:  Liquid::Template.parse(template.body_template).render(variables),
      email_subject: Liquid::Template.parse(
        template.email_subject_template
      ).render(variables),
      email_html: Liquid::Template.parse(
        template.email_html_template
      ).render(variables)
    }
  rescue ActiveRecord::RecordNotFound
    # フォールバック: 英語テンプレートを使用
    render(type: type, locale: 'en', variables: variables)
  end
end

# title_template
{{ follower_name }} さんがあなたをフォローしました
 
# body_template
{{ follower_name }} さんが {{ app_name }} であなたをフォローし始めました。
プロフィールを確認しましょう。
 
# email_html_template
<h1>新しいフォロワー</h1>
<p>{{ follower_name }} さんがあなたをフォローしました。</p>
<a href="{{ profile_url }}">プロフィールを見る</a>
<hr>
<p><a href="{{ unsubscribe_url }}">通知設定を変更する</a></p>

# app/models/scheduled_notification.rb
class ScheduledNotification < ApplicationRecord
  # scheduled_at: 送信予定時刻
  # timezone: ユーザーのタイムゾーン
  # status: pending / queued / cancelled
 
  scope :due, -> {
    where(status: 'pending')
      .where('scheduled_at <= ?', Time.current)
  }
end
 
# app/jobs/notification_scheduler_job.rb
class NotificationSchedulerJob < ApplicationJob
  queue_as :scheduler
 
  def perform
    ScheduledNotification.due.find_in_batches(batch_size: 1000) do |batch|
      batch.each do |scheduled|
        # アトミック更新で二重送信を防ぐ
        next unless ScheduledNotification
                      .where(id: scheduled.id, status: 'pending')
                      .update_all(status: 'queued') == 1
 
        NotificationService.send(
          user_id:  scheduled.user_id,
          type:     scheduled.notification_type,
          priority: scheduled.priority,
          channels: scheduled.channels,
          data:     scheduled.payload
        )
      end
    end
  end
end

// internal/worker/push_worker.go
package worker
 
import (
    "context"
    "encoding/json"
    "log/slog"
 
    "github.com/aws/aws-sdk-go-v2/service/sqs"
    "github.com/aws/aws-sdk-go-v2/service/sqs/types"
)
 
type PushMessage struct {
    NotificationID int64  `json:"notification_id"`
    UserID         int64  `json:"user_id"`
    Channel        string `json:"channel"`
    Title          string `json:"title"`
    Body           string `json:"body"`
    DeviceToken    string `json:"device_token"`
    Platform       string `json:"platform"` // "ios" or "android"
}
 
type PushWorker struct {
    sqsClient   *sqs.Client
    queueURL    string
    apnsClient  APNSClient
    fcmClient   FCMClient
    concurrency int
}
 
func (w *PushWorker) Run(ctx context.Context) error {
    sem := make(chan struct{}, w.concurrency)
 
    for {
        select {
        case <-ctx.Done():
            return nil
        default:
        }
 
        output, err := w.sqsClient.ReceiveMessage(ctx, &sqs.ReceiveMessageInput{
            QueueUrl:            &w.queueURL,
            MaxNumberOfMessages: 10,
            WaitTimeSeconds:     20, // Long polling
        })
        if err != nil {
            slog.Error("SQS receive error", "err", err)
            continue
        }
 
        for _, msg := range output.Messages {
            sem <- struct{}{}
            go func(m types.Message) {
                defer func() { <-sem }()
                w.processMessage(ctx, m)
            }(msg)
        }
    }
}
 
func (w *PushWorker) processMessage(ctx context.Context, msg types.Message) {
    var pm PushMessage
    if err := json.Unmarshal([]byte(*msg.Body), &pm); err != nil {
        slog.Error("unmarshal error", "err", err)
        return
    }
 
    var err error
    switch pm.Platform {
    case "ios":
        err = w.apnsClient.Send(ctx, APNSPayload{
            DeviceToken: pm.DeviceToken,
            Title:       pm.Title,
            Body:        pm.Body,
        })
    case "android":
        err = w.fcmClient.Send(ctx, FCMPayload{
            Token: pm.DeviceToken,
            Notification: FCMNotification{
                Title: pm.Title,
                Body:  pm.Body,
            },
        })
    }
 
    if err != nil {
        slog.Error("push delivery failed",
            "notification_id", pm.NotificationID,
            "platform", pm.Platform,
            "err", err,
        )
        // DLQへ（SQS の maxReceiveCount を超えると自動で移動）
        return
    }
 
    // 成功: メッセージを削除
    w.sqsClient.DeleteMessage(ctx, &sqs.DeleteMessageInput{
        QueueUrl:      &w.queueURL,
        ReceiptHandle: msg.ReceiptHandle,
    })
 
    slog.Info("push delivered",
        "notification_id", pm.NotificationID,
        "platform", pm.Platform,
    )
}

// internal/worker/email_worker.go
package worker
 
import (
    "context"
    "fmt"
 
    "github.com/aws/aws-sdk-go-v2/aws"
    "github.com/aws/aws-sdk-go-v2/service/ses"
    "github.com/aws/aws-sdk-go-v2/service/ses/types"
)
 
type EmailWorker struct {
    sesClient   *ses.Client
    fromAddress string
}
 
func (w *EmailWorker) SendEmail(ctx context.Context, msg EmailMessage) error {
    input := &ses.SendEmailInput{
        Source: aws.String(w.fromAddress),
        Destination: &types.Destination{
            ToAddresses: []string{msg.ToEmail},
        },
        Message: &types.Message{
            Subject: &types.Content{
                Data:    aws.String(msg.Subject),
                Charset: aws.String("UTF-8"),
            },
            Body: &types.Body{
                Html: &types.Content{
                    Data:    aws.String(msg.HtmlBody),
                    Charset: aws.String("UTF-8"),
                },
                Text: &types.Content{
                    Data:    aws.String(msg.TextBody),
                    Charset: aws.String("UTF-8"),
                },
            },
        },
        ConfigurationSetName: aws.String("notification-config"),
    }
 
    _, err := w.sesClient.SendEmail(ctx, input)
    if err != nil {
        return fmt.Errorf("SES send failed: %w", err)
    }
    return nil
}

# app/services/bounce_handler_service.rb
class BounceHandlerService
  # Amazon SES の SNS Webhook を受け取る
  def self.process(sns_payload)
    message = JSON.parse(sns_payload['Message'])
    bounce  = message['bounce']
    return unless bounce
 
    bounce['bouncedRecipients'].each do |recipient|
      email = recipient['emailAddress']
 
      case bounce['bounceType']
      when 'Permanent'
        # ハードバウンス: 即座に無効化
        EmailSuppression.create!(
          email:      email,
          reason:     'hard_bounce',
          bounced_at: Time.current
        )
      when 'Transient'
        # ソフトバウンス: 回数カウント
        record = EmailBounceRecord.find_or_create_by(email: email)
        record.increment!(:bounce_count)
 
        if record.bounce_count >= 3
          EmailSuppression.create!(
            email:      email,
            reason:     'soft_bounce_threshold',
            bounced_at: Time.current
          )
        end
      end
    end
  end
end
 
# 送信前にバウンスリストをチェック
class EmailNotificationWorker
  include Sidekiq::Worker
 
  def perform(notification_id)
    notification = Notification.find(notification_id)
    email = notification.user.email
 
    if EmailSuppression.exists?(email: email)
      notification.update!(status: 'suppressed')
      return
    end
 
    SES_CLIENT.send_email(
      destination:  { to_addresses: [email] },
      source:       'noreply@myapp.com',
      message: {
        subject: { data: notification.email_subject },
        body: {
          html: { data: notification.email_html_body },
          text: { data: notification.email_text_body }
        }
      },
      client_token: notification.idempotency_key
    )
 
    notification.update!(status: 'delivered', delivered_at: Time.current)
  end
end

# app/models/webhook_subscription.rb
class WebhookSubscription < ApplicationRecord
  # user_id, url, events (array), secret, status, failure_count
  encrypts :secret
 
  def self.notify(event_type:, payload:)
    active.where("? = ANY(events)", event_type).find_each do |sub|
      WebhookDeliveryJob.perform_later(
        subscription_id: sub.id,
        event_type:      event_type,
        payload:         payload
      )
    end
  end
end
 
# app/jobs/webhook_delivery_job.rb
class WebhookDeliveryJob < ApplicationJob
  queue_as :webhooks
  retry_on Net::TimeoutError, wait: :polynomially_longer, attempts: 5
 
  def perform(subscription_id:, event_type:, payload:)
    sub  = WebhookSubscription.find(subscription_id)
    body = payload.to_json
 
    # HMAC署名で受信側が真正性を検証できる
    signature = OpenSSL::HMAC.hexdigest('SHA256', sub.secret, body)
 
    response = HTTP.timeout(10)
      .headers(
        'Content-Type'        => 'application/json',
        'X-Webhook-Event'     => event_type,
        'X-Webhook-Signature' => "sha256=#{signature}"
      )
      .post(sub.url, body: body)
 
    unless response.status.success?
      sub.increment!(:failure_count)
      raise "Webhook failed: #{response.status} for #{sub.url}"
    end
 
    sub.update!(last_delivered_at: Time.current, failure_count: 0)
  end
end

# app/services/rate_limiter.rb
class RateLimiter
  # ユーザーごとのレート制限ルール
  LIMITS = {
    critical: { count: 10, window: 1.hour   },
    high:     { count: 20, window: 1.hour   },
    normal:   { count: 50, window: 24.hours },
    low:      { count: 5,  window: 24.hours }
  }.freeze
 
  def self.allow?(user_id:, priority:)
    rule = LIMITS[priority.to_sym]
    return true unless rule
 
    key   = "rate_limit:#{user_id}:#{priority}"
    count = redis.incr(key)
 
    # 初回のみ TTL をセット
    redis.expire(key, rule[:window].to_i) if count == 1
 
    count <= rule[:count]
  end
 
  private_class_method def self.redis
    Redis.current
  end
end

// Go版: Sliding Window でより精密なレート制限
// internal/ratelimit/limiter.go
package ratelimit
 
import (
    "context"
    "fmt"
    "time"
 
    "github.com/redis/go-redis/v9"
)
 
type Limiter struct {
    rdb *redis.Client
}
 
func (l *Limiter) Allow(ctx context.Context, userID int64, priority string) (bool, error) {
    limits := map[string]struct {
        count  int
        window time.Duration
    }{
        "critical": {10, time.Hour},
        "high":     {20, time.Hour},
        "normal":   {50, 24 * time.Hour},
        "low":      {5, 24 * time.Hour},
    }
 
    rule, ok := limits[priority]
    if !ok {
        return true, nil
    }
 
    key := fmt.Sprintf("rate_limit:%d:%s", userID, priority)
    now := time.Now()
    windowStart := now.Add(-rule.window).UnixMilli()
 
    pipe := l.rdb.Pipeline()
    pipe.ZRemRangeByScore(ctx, key, "-inf", fmt.Sprintf("%d", windowStart))
    pipe.ZCard(ctx, key)
    pipe.ZAdd(ctx, key, redis.Z{
        Score:  float64(now.UnixMilli()),
        Member: now.UnixNano(),
    })
    pipe.Expire(ctx, key, rule.window)
 
    results, err := pipe.Exec(ctx)
    if err != nil {
        return false, err
    }
 
    currentCount := results[1].(*redis.IntCmd).Val()
    return currentCount < int64(rule.count), nil
}

# app/services/deduplication_service.rb
class DeduplicationService
  DEDUP_WINDOW = 1.hour
 
  # true: 新規（送信OK）、false: 重複（スキップ）
  def self.check_and_mark(idempotency_key)
    redis.set(
      "notif:dedup:#{idempotency_key}",
      "1",
      nx: true,
      ex: DEDUP_WINDOW.to_i
    )
  end
 
  def self.generate_key(user_id:, type:, event_id:)
    Digest::SHA256.hexdigest("#{user_id}:#{type}:#{event_id}")
  end
 
  # Redisがダウンした場合のフォールバック
  def self.check_and_mark_with_fallback(idempotency_key)
    check_and_mark(idempotency_key)
  rescue Redis::CannotConnectError
    # DBレベルのユニーク制約にフォールバック
    Notification.create!(idempotency_key: idempotency_key)
    true
  rescue ActiveRecord::RecordNotUnique
    false
  end
end

# インフラ構成サマリー
 
メッセージキュー:
  SQS FIFO (Critical):
    VisibilityTimeout: 30s
    MessageRetentionPeriod: 4 days
    DLQ: sqs-notifications-critical-dlq
    maxReceiveCount: 3  # 3回失敗でDLQへ
 
  SQS Standard (Normal/Low):
    VisibilityTimeout: 120s
    DLQ: sqs-notifications-normal-dlq
    maxReceiveCount: 5
 
メール配信:
  Amazon SES:
    Dedicated IP: バウンス率改善
    Configuration Set: 開封/クリック追跡
    SNS Topic: ses-bounces → Lambda でバウンス処理
    SNS Topic: ses-complaints → 苦情処理
 
プッシュ通知:
  Amazon SNS:
    Platform Application: APNs / FCM を一元管理
    Endpoint ARN: デバイストークンと1対1対応
 
ストレージ:
  DynamoDB (通知ログ):
    PK: user_id
    SK: created_at#notification_id
    TTL: 90日でログを自動削除
    GSI: notification_type + status（集計用）
 
  ElastiCache Redis (Cluster Mode ON):
    用途1: 冪等性チェック（NX フラグ）
    用途2: レート制限（Sliding Window）
    用途3: デバイストークンキャッシュ
 
監視:
  CloudWatch Metrics:
    SQS ApproximateAgeOfOldestMessage: >= 60秒でアラーム
    SES BounceRate: >= 2% でアラーム
    SES ComplaintRate: >= 0.1% でアラーム
    DLQ MessageCount: >= 1 でPagerDuty通知
 
  Kinesis Data Firehose:
    通知イベントをS3へストリーミング
    Athena でアドホッククエリ（開封率・クリック率分析）

# app/models/notification_ab_test.rb
class NotificationAbTest < ApplicationRecord
  # id, notification_type, variant_a_template_id,
  # variant_b_template_id, traffic_split (0.5 = 50/50),
  # status (active/paused/completed), started_at, ended_at
 
  def self.assign_variant(user_id:, notification_type:)
    test = active.find_by(notification_type: notification_type)
    return :control unless test
 
    # ユーザーIDのハッシュで一貫したバリアント割り当て
    # 同じユーザーは常に同じバリアントを受け取る
    hash = Digest::CRC32.checksum("#{test.id}:#{user_id}")
    fraction = (hash % 100) / 100.0
 
    fraction < test.traffic_split ? :variant_a : :variant_b
  end
end
 
# 開封追跡（1px トラッキングピクセル）
class EmailTrackingController < ApplicationController
  skip_before_action :authenticate_user!
 
  def open
    event = EmailEvent.find_by!(token: params[:token])
    event.update!(opened_at: Time.current) unless event.opened_at
 
    # 1x1 透明GIF を返す
    send_data "\x47\x49\x46\x38\x39\x61\x01\x00\x01\x00".b,
              type: 'image/gif', disposition: 'inline'
  end
 
  def click
    event = EmailEvent.find_by!(token: params[:token])
    event.update!(clicked_at: Time.current, clicked_url: params[:url])
 
    redirect_to params[:url], allow_other_host: true
  end
end

// internal/metrics/notification_metrics.go
package metrics
 
import (
    "github.com/prometheus/client_golang/prometheus"
    "github.com/prometheus/client_golang/prometheus/promauto"
)
 
var (
    NotificationsSent = promauto.NewCounterVec(prometheus.CounterOpts{
        Name: "notifications_sent_total",
        Help: "Total notifications sent",
    }, []string{"channel", "priority", "status"})
 
    NotificationLatency = promauto.NewHistogramVec(prometheus.HistogramOpts{
        Name:    "notification_delivery_latency_seconds",
        Help:    "Notification delivery latency",
        Buckets: []float64{0.1, 0.5, 1, 5, 10, 30, 60},
    }, []string{"channel", "priority"})
 
    BounceRate = promauto.NewGaugeVec(prometheus.GaugeOpts{
        Name: "email_bounce_rate",
        Help: "Email bounce rate (rolling 24h)",
    }, []string{"bounce_type"})
 
    QueueDepth = promauto.NewGaugeVec(prometheus.GaugeOpts{
        Name: "notification_queue_depth",
        Help: "Number of messages in queue",
    }, []string{"queue", "priority"})
)

# app/models/user_notification_setting.rb
class UserNotificationSetting < ApplicationRecord
  # GDPR: 明示的なオプトインを記録
  # opted_in_at, opt_in_ip, opt_in_user_agent
 
  scope :globally_opted_out, -> { where(global_opt_out: true) }
 
  def globally_opted_out?
    global_opt_out
  end
 
  def channel_enabled?(channel)
    preferences.fetch(channel.to_s, true)
  end
 
  # 特定電子メール法（日本）: 受信拒否を即時反映
  def opt_out_email!
    update!(
      email_notifications: false,
      email_opted_out_at:  Time.current
    )
  end
end
 
# CAN-SPAM 準拠: メールフッターに配信停止リンク必須
# app/controllers/unsubscribes_controller.rb
class UnsubscribesController < ApplicationController
  skip_before_action :authenticate_user!
 
  def create
    user_id = UnsubscribeToken.decode!(params[:token])
    UserNotificationSetting
      .find_or_create_by(user_id: user_id)
      .update!(global_opt_out: true, opted_out_at: Time.current)
 
    render :success
  rescue JWT::DecodeError
    render :invalid_token, status: :bad_request
  end
end

# ECS Auto Scaling 設定
notification-push-worker:
  desired: 10
  min: 5
  max: 100
  scaling_policy:
    metric: SQS_ApproximateNumberOfMessagesVisible
    target_value: 1000  # キューに1000件溜まったらスケールアウト
    scale_out_cooldown: 60
    scale_in_cooldown: 300
 
notification-email-worker:
  desired: 5
  min: 2
  max: 50
  scaling_policy:
    metric: SQS_ApproximateNumberOfMessagesVisible
    target_value: 500

// APNs / FCM への接続プールを再利用する
// internal/apns/connection_pool.go
package apns
 
import "sync"
 
type ConnectionPool struct {
    mu      sync.Mutex
    conns   []*Connection
    maxSize int
    idx     int
}
 
func (p *ConnectionPool) Get() *Connection {
    p.mu.Lock()
    defer p.mu.Unlock()
 
    conn := p.conns[p.idx%len(p.conns)]
    p.idx++
    return conn
}