データベースの選択 — RDB以外の選択肢

# Gemfile
gem 'aws-sessionstore-dynamodb'

# DynamoDB テーブル作成（TTL自動削除付き）
aws dynamodb create-table \
  --table-name buzz-sessions \
  --attribute-definitions AttributeName=session_id,AttributeType=S \
  --key-schema AttributeName=session_id,KeyType=HASH \
  --billing-mode PAY_PER_REQUEST \
  --region ap-northeast-1
 
# TTL属性を有効化（期限切れセッションを自動削除）
aws dynamodb update-time-to-live \
  --table-name buzz-sessions \
  --time-to-live-specification "Enabled=true, AttributeName=expires_at"

# config/initializers/session_store.rb
Rails.application.config.session_store(
  :dynamo_db_store,
  table_name: 'buzz-sessions',
  table_hash_key: 'session_id',
  ttl: 30.days.to_i,            # セッション有効期限
  ttl_attribute: 'expires_at',  # TTLとして使うカラム名
  region: 'ap-northeast-1',
  pool:   10  # 接続プールサイズ
)

# app/services/activity_feed_service.rb
class ActivityFeedService
  DYNAMODB = Aws::DynamoDB::Client.new(region: 'ap-northeast-1')
  TABLE_NAME = 'buzz-activity-feed'
 
  # DynamoDBのテーブル設計:
  # PK (パーティションキー): recipient_id（誰が受け取るか）
  # SK (ソートキー):         activity_id（SnowflakeID、時系列でソートできる）
  # 追加属性:                actor_id, action, target_type, target_id, created_at, ttl
 
  # アクティビティを記録（全受信者のフィードに書き込む）
  def self.record(actor_id:, action:, target_type:, target_id:, recipient_ids:)
    activity_id = Buzz::SnowflakeId.generate.to_s
 
    # 最大25件ずつバッチ書き込み
    recipient_ids.each_slice(25) do |batch_ids|
      requests = batch_ids.map do |recipient_id|
        {
          put_request: {
            item: {
              'recipient_id' => recipient_id.to_s,
              'activity_id'  => activity_id,
              'actor_id'     => actor_id.to_s,
              'action'       => action,        # 'like', 'follow', 'comment', 'mention'
              'target_type'  => target_type,   # 'Post', 'User'
              'target_id'    => target_id.to_s,
              'created_at'   => Time.current.iso8601,
              'ttl'          => (Time.current + 90.days).to_i  # 90日後に自動削除
            }
          }
        }
      end
 
      DYNAMODB.batch_write_item(
        request_items: { TABLE_NAME => requests }
      )
    end
  end
 
  # フィードを取得（最新N件）
  def self.get_feed(user_id:, limit: 20, next_token: nil)
    params = {
      table_name: TABLE_NAME,
      key_condition_expression: 'recipient_id = :uid',
      expression_attribute_values: { ':uid' => user_id.to_s },
      scan_index_forward: false,  # 降順（最新から）
      limit: limit
    }
 
    # ページネーション
    params[:exclusive_start_key] = JSON.parse(next_token) if next_token
 
    result = DYNAMODB.query(params)
 
    {
      items:      enrich_activities(result.items),
      next_token: result.last_evaluated_key&.to_json
    }
  end
 
  private
 
  def self.enrich_activities(items)
    # actor_idからユーザー情報をバッチ取得（Aurora）
    actor_ids = items.map { |i| i['actor_id'].to_i }.uniq
    actors = User.where(id: actor_ids)
                 .select(:id, :username, :avatar_url)
                 .index_by { |u| u.id.to_s }
 
    items.map do |item|
      actor = actors[item['actor_id']]
      {
        activity_id: item['activity_id'],
        actor: {
          id:         actor&.id,
          username:   actor&.username,
          avatar_url: actor&.avatar_url
        },
        action:      item['action'],
        target_type: item['target_type'],
        target_id:   item['target_id'].to_i,
        created_at:  item['created_at']
      }
    end
  end
end

# GSI（グローバルセカンダリインデックス）を追加
aws dynamodb update-table \
  --table-name buzz-activity-feed \
  --global-secondary-index-updates '[
    {
      "Create": {
        "IndexName": "actor-index",
        "KeySchema": [
          {"AttributeName": "actor_id", "KeyType": "HASH"},
          {"AttributeName": "activity_id", "KeyType": "RANGE"}
        ],
        "Projection": {"ProjectionType": "ALL"},
        "ProvisionedThroughput": {"ReadCapacityUnits": 5, "WriteCapacityUnits": 5}
      }
    }
  ]'

# Gemfile
gem 'opensearch-ruby'
gem 'searchkick'  # Rails との統合を簡単にする

# config/initializers/searchkick.rb
Searchkick.client = OpenSearch::Client.new(
  url: ENV['OPENSEARCH_URL'],
  transport_options: {
    ssl: { verify: Rails.env.production? }
  },
  retry_on_failure: 3,
  log: Rails.env.development?
)
 
# インデックス設定のデフォルト
Searchkick.timeout = 10
Searchkick.queue_name = :search_index  # Sidekiqのキュー名

# app/models/post.rb
class Post < ApplicationRecord
  searchkick(
    # 検索対象のフィールド設定
    text_start:  [:content, :hashtags],
    highlight:   [:content],
    word_middle: [:content],
 
    # インデックス名（バージョン付き）
    index_name: "buzz_posts_#{Rails.env}",
 
    # 日本語解析設定
    settings: {
      analysis: {
        tokenizer: {
          kuromoji: {
            type: 'kuromoji_tokenizer',
            mode: 'search'  # 検索モード（より細かくトークン化）
          }
        },
        analyzer: {
          japanese: {
            type:      'custom',
            tokenizer: 'kuromoji',
            filter:    ['kuromoji_baseform', 'lowercase', 'cjk_width', 'kuromoji_stemmer']
          },
          japanese_index: {
            type:      'custom',
            tokenizer: 'kuromoji',
            filter:    ['kuromoji_baseform', 'lowercase', 'cjk_width']
          }
        }
      }
    },
 
    # マッピング設定
    mappings: {
      properties: {
        content: {
          type:            'text',
          analyzer:        'japanese_index',
          search_analyzer: 'japanese'
        },
        hashtags: {
          type: 'keyword'  # 完全一致検索
        },
        likes_count: {
          type: 'long'
        },
        created_at: {
          type: 'date'
        }
      }
    }
  )
 
  # OpenSearchに保存するデータ
  def search_data
    {
      content:     content,
      hashtags:    extract_hashtags,
      user_id:     user_id,
      username:    user.username,     # デノーマライズ（JOIN不要にする）
      likes_count: likes_count,
      created_at:  created_at,
      is_deleted:  deleted_at.present?
    }
  end
 
  # 削除されたらインデックスからも削除
  after_commit :reindex_async
 
  private
 
  def extract_hashtags
    content.scan(/#[\w぀-ゟ゠-ヿ一-龯]+/)
           .map(&:downcase)
           .uniq
  end
end

# 検索の実装
class SearchController < ApplicationController
  def index
    query    = params[:q].to_s.strip
    page     = params[:page]&.to_i || 1
    per_page = 20
 
    @results = Post.search(
      query.presence || '*',  # クエリが空の場合は全件
 
      fields: [
        'content^3',   # contentフィールドのスコアを3倍に重み付け
        'hashtags^2',  # hashtagsフィールドは2倍
        'username^1'
      ],
 
      highlight: {
        fields: {
          content: {
            fragment_size:       150,
            number_of_fragments: 2,
            pre_tag:  '<mark>',
            post_tag: '</mark>'
          }
        }
      },
 
      where: {
        is_deleted: false,
        created_at: { gte: 6.months.ago },  # 直近6ヶ月に絞る
        _or: [
          { likes_count: { gte: 5 } },  # いいね5件以上
          { user_id: current_user.following_ids.first(100) }  # フォロー中ユーザー
        ]
      },
 
      order: {
        _score:      :desc,    # 関連度スコア降順
        likes_count: :desc     # 同スコアならいいね数が多い順
      },
 
      page:     page,
      per_page: per_page,
 
      # タイポ（誤字）を許容
      misspellings: {
        below:     5,    # 5文字以下はタイポ許容しない
        edit_distance: 1 # 編集距離1まで（1文字違い）
      },
 
      # 検索キーワードの提案（もしかして？）
      suggest: query.present?
    )
  end
end

# インデックスの同期（非同期）
class SearchIndexJob < ApplicationJob
  queue_as :search_index
  discard_on ActiveRecord::RecordNotFound
 
  def perform(post_id)
    post = Post.find_by(id: post_id)
 
    if post
      post.reindex
    else
      # 削除された投稿はインデックスから削除
      Post.search_index.remove("post_#{post_id}")
    end
  end
end

# app/services/rate_limiter_service.rb
class RateLimiterService
  REDIS = Redis.new(url: ENV['REDIS_URL'])
 
  # スライディングウィンドウ方式のレート制限
  # 「直近60秒間に10回まで」という制限を正確に実装
  def self.allow?(identifier:, limit:, window: 60)
    key    = "rate_limit:#{identifier}"
    now    = Time.current.to_f
    window_start = now - window
 
    result = REDIS.multi do |pipeline|
      # 古いリクエスト記録を削除（ウィンドウ外）
      pipeline.zremrangebyscore(key, '-inf', window_start)
 
      # 現在のウィンドウ内のリクエスト数
      pipeline.zcard(key)
 
      # 今回のリクエストを記録
      pipeline.zadd(key, now, "#{now}-#{SecureRandom.hex(4)}")
 
      # TTLを設定
      pipeline.expire(key, window + 1)
    end
 
    count = result[1]
    count < limit  # limit未満なら許可
  end
 
  # 残り回数を返す
  def self.remaining(identifier:, limit:, window: 60)
    key   = "rate_limit:#{identifier}"
    now   = Time.current.to_f
    window_start = now - window
 
    REDIS.zremrangebyscore(key, '-inf', window_start)
    current = REDIS.zcard(key)
    [limit - current, 0].max
  end
end

# app/controllers/api/v1/posts_controller.rb
module Api
  module V1
    class PostsController < ApiController
      before_action :check_rate_limit, only: [:create]
 
      def create
        @post = PostCreator.new(user: current_user, params: post_params).call
        render json: PostSerializer.new(@post).to_json, status: :created
      end
 
      private
 
      def check_rate_limit
        identifier = "user:#{current_user.id}:posts"
        remaining  = RateLimiterService.remaining(identifier: identifier, limit: 20, window: 60)
 
        # ヘッダーにレート制限情報を返す（APIクライアントが参照できる）
        response.headers['X-RateLimit-Limit']     = '20'
        response.headers['X-RateLimit-Remaining'] = remaining.to_s
        response.headers['X-RateLimit-Reset']     = (Time.current + 60).to_i.to_s
 
        unless RateLimiterService.allow?(identifier: identifier, limit: 20, window: 60)
          render json: {
            error:       'rate_limit_exceeded',
            message:     '1分間に投稿できるのは20件までです',
            retry_after: 60
          }, status: :too_many_requests
        end
      end
 
      def post_params
        params.require(:post).permit(:content, :image)
      end
    end
  end
end

# app/services/activity_feed_service.rb（フォールバック付き）
class ActivityFeedService
  def self.get_feed(user_id:, limit: 20, next_token: nil)
    # DynamoDBが使えない場合はAuroraにフォールバック
    Timeout::timeout(3) do  # 3秒でタイムアウト
      get_feed_from_dynamodb(user_id: user_id, limit: limit, next_token: next_token)
    end
  rescue Aws::DynamoDB::Errors::ServiceError,
         Timeout::Error => e
    Rails.logger.error "DynamoDB error, falling back to Aurora: #{e.message}"
    Sentry.capture_exception(e, level: :warning)
    get_feed_from_aurora(user_id: user_id, limit: limit)
  end
 
  private
 
  def self.get_feed_from_aurora(user_id:, limit:)
    # Auroraからアクティビティを取得（遅いがフォールバックとして機能）
    Notification.where(recipient_id: user_id)
                .order(created_at: :desc)
                .limit(limit)
                .includes(:actor, :target)
                .map { |n| serialize_notification(n) }
  end
end

# app/services/search_service.rb（フォールバック付き）
class SearchService
  def self.search(query:, page: 1, per: 20)
    # OpenSearchが使えない場合はPostgreSQLにフォールバック
    Post.search(query, page: page, per_page: per)
  rescue Faraday::ConnectionFailed,
         OpenSearch::Transport::Transport::Errors::ServiceUnavailable => e
    Rails.logger.error "OpenSearch error, falling back to PostgreSQL: #{e.message}"
    search_with_postgres(query: query, page: page, per: per)
  end
 
  private
 
  def self.search_with_postgres(query:, page:, per:)
    # pg_trgm を使った全文検索（OpenSearchより遅いが動く）
    Post.search_by_content(query)
        .page(page).per(per)
  end
end

# Outboxパターン: DBとメッセージキューの一貫性を保つ
# 投稿作成時: AuroraとDynamoDBとOpenSearchを同期
 
class PostCreator
  def initialize(user:, params:)
    @user   = user
    @params = params
  end
 
  def call
    post = nil
 
    # 1. Aurora に投稿を保存（これが唯一の信頼できるソース）
    ActiveRecord::Base.transaction do
      post = Post.create!(
        user:    @user,
        content: @params[:content]
      )
 
      # Outboxテーブルに「OpenSearchへの同期が必要」と記録
      OutboxEvent.create!(
        aggregate_type: 'Post',
        aggregate_id:   post.id,
        event_type:     'PostCreated',
        payload:        post.as_json
      )
    end
 
    # 2. DynamoDBへの同期（非同期）
    SyncToDynamoJob.perform_later('Post', post.id, 'created')
 
    # 3. OpenSearchへの同期（非同期、Outboxから）
    SearchIndexJob.perform_later(post.id)
 
    post
  end
end
 
# Outboxテーブル（Aurora内）
class OutboxEvent < ApplicationRecord
  # id, aggregate_type, aggregate_id, event_type, payload, processed_at
  scope :pending, -> { where(processed_at: nil) }
end