設計問題: 検索エンジン — 全文検索の仕組み

// crawler/worker.go
package crawler
 
import (
	"context"
	"log"
	"net/http"
	"net/url"
	"sync"
	"time"
 
	"golang.org/x/time/rate"
)
 
// ドメインごとのレートリミッター管理
type DomainLimiter struct {
	mu       sync.Mutex
	limiters map[string]*rate.Limiter
}
 
func NewDomainLimiter() *DomainLimiter {
	return &DomainLimiter{
		limiters: make(map[string]*rate.Limiter),
	}
}
 
func (dl *DomainLimiter) GetLimiter(domain string) *rate.Limiter {
	dl.mu.Lock()
	defer dl.mu.Unlock()
	if limiter, ok := dl.limiters[domain]; ok {
		return limiter
	}
	// 1ドメインあたり1req/secのレート制限
	limiter := rate.NewLimiter(rate.Every(time.Second), 1)
	dl.limiters[domain] = limiter
	return limiter
}
 
type CrawlerWorker struct {
	client      *http.Client
	domainLimit *DomainLimiter
	robotsCache *RobotsCache
	s3Client    S3Client
	sqsClient   SQSClient
}
 
func NewCrawlerWorker(s3 S3Client, sqs SQSClient) *CrawlerWorker {
	return &CrawlerWorker{
		client:      &http.Client{Timeout: 30 * time.Second},
		domainLimit: NewDomainLimiter(),
		robotsCache: NewRobotsCache(),
		s3Client:    s3,
		sqsClient:   sqs,
	}
}
 
func (w *CrawlerWorker) ProcessURL(ctx context.Context, rawURL string) error {
	parsed, err := url.Parse(rawURL)
	if err != nil {
		return err
	}
	domain := parsed.Host
 
	// robots.txtチェック
	if w.robotsCache.IsDisallowed(domain, rawURL) {
		log.Printf("robots.txt disallows: %s", rawURL)
		return nil
	}
 
	// ドメインごとのレートリミット適用
	limiter := w.domainLimit.GetLimiter(domain)
	if err := limiter.Wait(ctx); err != nil {
		return err
	}
 
	// HTML取得
	resp, err := w.client.Get(rawURL)
	if err != nil {
		return err
	}
	defer resp.Body.Close()
 
	if resp.StatusCode != http.StatusOK {
		return nil
	}
 
	// S3に保存 + 解析ジョブをSQSへ
	key := urlToS3Key(rawURL)
	if err := w.s3Client.PutObject(ctx, key, resp.Body); err != nil {
		return err
	}
	return w.sqsClient.SendMessage(ctx, ParseJob{URL: rawURL, S3Key: key})
}
 
// 並列ワーカープール
func RunWorkerPool(ctx context.Context, concurrency int, urls <-chan string) {
	worker := NewCrawlerWorker(newS3Client(), newSQSClient())
	var wg sync.WaitGroup
 
	for i := 0; i < concurrency; i++ {
		wg.Add(1)
		go func() {
			defer wg.Done()
			for rawURL := range urls {
				if err := worker.ProcessURL(ctx, rawURL); err != nil {
					log.Printf("crawl error %s: %v", rawURL, err)
				}
			}
		}()
	}
	wg.Wait()
}

// frontier/priority_queue.go
package frontier
 
import (
	"context"
	"encoding/json"
 
	"github.com/aws/aws-sdk-go-v2/service/sqs"
)
 
type URLJob struct {
	URL      string  `json:"url"`
	Priority float64 `json:"priority"` // 0.0〜1.0（PageRankスコア等）
	Depth    int     `json:"depth"`
}
 
type PriorityFrontier struct {
	sqsClient      *sqs.Client
	highQueueURL   string // Priority 0.7以上
	mediumQueueURL string // Priority 0.3〜0.7
	lowQueueURL    string // Priority 0.3未満
}
 
func (f *PriorityFrontier) Enqueue(ctx context.Context, job URLJob) error {
	queueURL := f.selectQueue(job.Priority)
	body, _ := json.Marshal(job)
	_, err := f.sqsClient.SendMessage(ctx, &sqs.SendMessageInput{
		QueueUrl:    &queueURL,
		MessageBody: strPtr(string(body)),
	})
	return err
}
 
func (f *PriorityFrontier) selectQueue(priority float64) string {
	switch {
	case priority >= 0.7:
		return f.highQueueURL
	case priority >= 0.3:
		return f.mediumQueueURL
	default:
		return f.lowQueueURL
	}
}
 
func strPtr(s string) *string { return &s }

# app/services/robots_parser.rb
class RobotsParser
  def initialize(domain)
    @rules = parse_robots_txt(domain)
  end
 
  def disallows?(url)
    @rules.any? do |rule|
      rule[:agent] == '*' && url.match?(rule[:pattern])
    end
  end
 
  def crawl_delay
    @rules.find { |r| r[:type] == :crawl_delay }&.dig(:value) || 1
  end
 
  private
 
  def parse_robots_txt(domain)
    response = HTTParty.get("https://#{domain}/robots.txt", timeout: 10)
    return [] unless response.success?
 
    rules = []
    current_agent = nil
    response.body.each_line do |line|
      line = line.strip.downcase
      if line.start_with?('user-agent:')
        current_agent = line.split(':', 2).last.strip
      elsif line.start_with?('disallow:') && current_agent == '*'
        path = line.split(':', 2).last.strip
        rules << { agent: '*', pattern: Regexp.new("^#{Regexp.escape(path)}"), type: :disallow }
      elsif line.start_with?('crawl-delay:')
        delay = line.split(':', 2).last.strip.to_i
        rules << { type: :crawl_delay, value: delay }
      end
    end
    rules
  end
end

// indexer/compression.go
package indexer
 
import "encoding/binary"
 
// Delta Encoding: doc_idの差分を格納してサイズを削減
// 例: [101, 307, 892] → [101, 206, 585]（差分）
// さらにVarInt（可変長整数）でエンコードすると70%以上圧縮できる
 
func EncodePostingList(docIDs []uint64) []byte {
	buf := make([]byte, 0, len(docIDs)*4)
	var prev uint64
	for _, id := range docIDs {
		delta := id - prev
		buf = appendVarInt(buf, delta)
		prev = id
	}
	return buf
}
 
func appendVarInt(buf []byte, v uint64) []byte {
	var tmp [binary.MaxVarintLen64]byte
	n := binary.PutUvarint(tmp[:], v)
	return append(buf, tmp[:n]...)
}
 
func DecodePostingList(data []byte) []uint64 {
	var ids []uint64
	var prev uint64
	for len(data) > 0 {
		delta, n := binary.Uvarint(data)
		data = data[n:]
		id := prev + delta
		ids = append(ids, id)
		prev = id
	}
	return ids
}

# app/services/japanese_tokenizer.rb
# Kuromojiを使った日本語形態素解析（natto gem経由でMeCabを利用）
require 'natto'
 
class JapaneseTokenizer
  STOP_WORDS = %w[の は が を に で も と や].freeze
 
  def self.tokenize(text)
    nm = Natto::MeCab.new
    tokens = []
 
    nm.parse(text) do |node|
      next if node.is_eos? || node.is_bos?
 
      feature = node.feature.split(',')
      part_of_speech = feature[0]  # 品詞（名詞、動詞など）
      base_form      = feature[6]  # 基本形（原形）
 
      # 名詞・動詞・形容詞のみ抽出（ストップワード除外）
      next unless %w[名詞 動詞 形容詞].include?(part_of_speech)
      next if STOP_WORDS.include?(node.surface)
      next if base_form == '*'
 
      tokens << base_form.downcase
    end
 
    tokens
  end
 
  # 英語はステミング（語幹化）で対応
  def self.stem_english(word)
    word.gsub(/ing$/, '').gsub(/ed$/, '').gsub(/s$/, '')
  end
 
  # 日本語と英語が混在するテキストを統合処理
  def self.tokenize_mixed(text)
    japanese_parts = text.scan(/[\p{Hiragana}\p{Katakana}\p{Han}]+/)
    english_parts  = text.scan(/[a-zA-Z]+/)
 
    japanese_tokens = japanese_parts.flat_map { |p| tokenize(p) }
    english_tokens  = english_parts.map { |w| stem_english(w.downcase) }
 
    (japanese_tokens + english_tokens).uniq.compact
  end
end

# Gemfile
gem 'pg_search'
 
# app/models/document.rb
class Document < ApplicationRecord
  include PgSearch::Model
 
  # PostgreSQLのtsvector型を使った全文検索
  pg_search_scope :search_fulltext,
    against: {
      title:   'A',  # タイトルは重み高
      body:    'B',  # 本文は重み中
      summary: 'C'   # サマリーは重み低
    },
    using: {
      tsearch: {
        dictionary: 'english',
        tsvector_column: 'searchable_content',
        highlight: {
          StartSel: '<mark>',
          StopSel:  '</mark>',
          MaxWords: 50,
          MinWords: 15,
          ShortWord: 3
        }
      }
    }
end
 
# db/migrate/add_search_index_to_documents.rb
class AddSearchIndexToDocuments < ActiveRecord::Migration[7.1]
  def change
    add_column :documents, :searchable_content, :tsvector
    add_index  :documents, :searchable_content, using: :gin
 
    execute <<~SQL
      CREATE OR REPLACE FUNCTION update_searchable_content()
      RETURNS trigger AS $$
      BEGIN
        NEW.searchable_content :=
          setweight(to_tsvector('english', coalesce(NEW.title, '')), 'A') ||
          setweight(to_tsvector('english', coalesce(NEW.body, '')), 'B');
        RETURN NEW;
      END;
      $$ LANGUAGE plpgsql;
 
      CREATE TRIGGER documents_searchable_update
      BEFORE INSERT OR UPDATE ON documents
      FOR EACH ROW EXECUTE FUNCTION update_searchable_content();
    SQL
  end
end
 
# app/controllers/search_controller.rb
class SearchController < ApplicationController
  def index
    query = params[:q].to_s.strip
    return render json: { results: [] } if query.blank?
 
    cache_key = "search:#{Digest::SHA256.hexdigest(query)}:#{params[:page]}"
    results = Rails.cache.fetch(cache_key, expires_in: 5.minutes) do
      Document.search_fulltext(query)
              .with_pg_search_highlight
              .order(pg_search_rank: :desc)
              .page(params[:page]).per(10)
              .map { |d| serialize_result(d) }
    end
 
    render json: { results: results, query: query }
  end
 
  private
 
  def serialize_result(doc)
    {
      id:      doc.id,
      url:     doc.url,
      title:   doc.title,
      snippet: doc.pg_search_highlight,
      score:   doc.pg_search_rank
    }
  end
end

# app/services/ranking_service.rb
class RankingService
  # 検索結果のスコア計算（複数シグナルの統合）
  def self.calculate_score(doc_id, query_terms, user_context)
    bm25_score      = calculate_bm25(doc_id, query_terms)
    page_rank       = PageRankCache.get(doc_id) || 0.0
    freshness_score = calculate_freshness(doc_id)
    personal_score  = calculate_personalization(doc_id, user_context)
    quality_score   = calculate_quality(doc_id)  # スパム判定、広告比率
 
    bm25_score      * 0.35 +
      page_rank     * 0.30 +
      freshness_score * 0.15 +
      personal_score  * 0.10 +
      quality_score   * 0.10
  end
 
  def self.calculate_bm25(doc_id, query_terms, k1: 1.5, b: 0.75)
    doc    = DocumentMetadata.find(doc_id)
    avg_dl = DocumentMetadata.average(:token_count).to_f
 
    query_terms.sum do |term|
      tf  = doc.term_frequency(term).to_f
      idf = Math.log((TOTAL_DOCS - doc_frequency(term) + 0.5) /
                     (doc_frequency(term) + 0.5) + 1)
      idf * (tf * (k1 + 1)) /
        (tf + k1 * (1 - b + b * doc.token_count / avg_dl))
    end
  end
 
  def self.calculate_freshness(doc_id)
    last_crawled = CrawlMetadata.last_crawled_at(doc_id)
    days_old = (Time.current - last_crawled) / 1.day
    Math.exp(-0.01 * days_old)  # 指数減衰：古いほど低スコア
  end
 
  def self.calculate_personalization(doc_id, user_context)
    return 0.0 unless user_context[:user_id]
    domain = DocumentMetadata.domain(doc_id)
    UserDomainAffinity.score(user_context[:user_id], domain)
  end
end

# app/services/query_processor.rb
class QueryProcessor
  def self.process(raw_query, user_id: nil)
    # 1. クエリの正規化
    normalized = normalize(raw_query)
 
    # 2. スペルミス訂正
    corrected, was_corrected = correct_spelling(normalized)
 
    # 3. クエリの解析（AND/OR/NOT検索）
    parsed = parse_query(corrected)
 
    # 4. 検索実行
    results = execute_search(parsed, user_id: user_id)
 
    # 5. スニペット生成
    results_with_snippets = attach_snippets(results, corrected)
 
    {
      results:         results_with_snippets,
      corrected_query: was_corrected ? corrected : nil,
      total:           results.total_count
    }
  end
 
  # 編集距離（Levenshtein）を使ったスペルミス訂正
  def self.correct_spelling(query)
    tokens    = query.split
    corrected = tokens.map do |token|
      candidates = SpellDictionary.nearest_words(token, max_distance: 2)
      candidates.max_by { |w| DictionaryFrequency.get(w) } || token
    end
    was_corrected = corrected != tokens
    [corrected.join(' '), was_corrected]
  end
 
  # スニペット生成（検索語周辺の文脈を抽出してハイライト）
  def self.generate_snippet(doc_content, query_terms, window: 80)
    sentences = doc_content.split(/[。.!?]/)
    best = sentences.max_by { |s| query_terms.count { |t| s.include?(t) } }
    return '' unless best
 
    pos   = query_terms.filter_map { |t| best.index(t) }.min || 0
    start = [pos - window / 2, 0].max
    snippet = best[start, window]
 
    query_terms.reduce(snippet) do |s, term|
      s.gsub(term, "<mark>#{term}</mark>")
    end
  end
end

// indexer/kafka_consumer.go
package indexer
 
import (
	"context"
	"encoding/json"
	"log"
 
	"github.com/segmentio/kafka-go"
)
 
type ParsedDocument struct {
	URL       string   `json:"url"`
	Title     string   `json:"title"`
	Body      string   `json:"body"`
	Tokens    []string `json:"tokens"`
	CrawledAt string   `json:"crawled_at"`
}
 
type IndexerConsumer struct {
	reader    *kafka.Reader
	osClient  OpenSearchClient
	batchSize int
}
 
func NewIndexerConsumer(brokers []string, os OpenSearchClient) *IndexerConsumer {
	return &IndexerConsumer{
		reader: kafka.NewReader(kafka.ReaderConfig{
			Brokers:  brokers,
			Topic:    "parsed-doc",
			GroupID:  "indexer-group",
			MinBytes: 10e3,
			MaxBytes: 10e6,
		}),
		osClient:  os,
		batchSize: 500,
	}
}
 
func (c *IndexerConsumer) Run(ctx context.Context) error {
	batch := make([]ParsedDocument, 0, c.batchSize)
 
	for {
		msg, err := c.reader.ReadMessage(ctx)
		if err != nil {
			return err
		}
 
		var doc ParsedDocument
		if err := json.Unmarshal(msg.Value, &doc); err != nil {
			log.Printf("parse error: %v", err)
			continue
		}
 
		batch = append(batch, doc)
 
		// バッチが溜まったらBulk APIでまとめてインデックス
		if len(batch) >= c.batchSize {
			if err := c.osClient.BulkIndex(ctx, batch); err != nil {
				log.Printf("bulk index error: %v", err)
			} else {
				log.Printf("indexed %d documents", len(batch))
			}
			batch = batch[:0]
		}
	}
}

クロールシステム:
  EC2クローラー群:
    インスタンスタイプ: c6i.2xlarge（スポット）
    台数: 10台（オートスケーリング、最大30台）
    コスト削減: スポット価格でオンデマンドの70%削減
  SQS URL Frontier:
    優先度別に3キュー（High / Medium / Low Priority）
    Visibility Timeout: 60秒
    Dead Letter Queue: 5回失敗したURLを隔離
  S3生HTML保存:
    ライフサイクル: 90日後にGlacierへ移行
    ストレージクラス: Intelligent-Tiering
 
解析・インデックスパイプライン:
  Amazon MSK（Kafka）:
    トピック: raw-html（パーティション: 24）、parsed-doc（パーティション: 24）
    レプリカ: 3
    保持期間: 7日間
  ECS Parser（Fargate）:
    タスク数: 20（QPSに応じてオートスケール）
    CPU: 2vCPU、メモリ: 4GB
  ECS Indexer（Fargate）:
    タスク数: 10
    OpenSearch Bulk APIで500件ずつバッチ処理
 
検索インフラ:
  Amazon OpenSearch Service:
    バージョン: OpenSearch 2.x
    マスターノード: 3台（m6g.large.search）
    データノード: 20台（r6g.2xlarge.search、192GB SSD）
    シャード設計: 1,000シャード（5B docs / 5M per shard）
    レプリカ: 1（可用性確保）
  ElastiCache for Redis:
    クラスターモード: 3シャード × 2レプリカ
    用途: 検索結果キャッシュ（Top-10,000クエリ）
    TTL: 5分
  API Gateway + Lambda:
    検索APIのスケールアウト対応
    Provisioned Concurrency: 100（コールドスタート対策）
 
ストレージ:
  DynamoDB: URLメタデータ（クロール日時、URLハッシュ、HTTP Status）
  Aurora PostgreSQL: PageRankスコア、インデックスメタデータ

# app/services/personalization_service.rb
class PersonalizationService
  HISTORY_WINDOW = 30.days
 
  # ユーザーのクリック履歴からドメイン親和性を更新
  def self.record_click(user_id:, url:, query:, rank:)
    domain = URI.parse(url).host
    # 上位に表示されてもクリックされた場合に重みを増やす
    click_weight = rank <= 3 ? 1.5 : 1.0
 
    Rails.cache.write(
      "affinity:#{user_id}:#{domain}",
      current_affinity(user_id, domain) + click_weight,
      expires_in: HISTORY_WINDOW
    )
 
    SearchLog.create!(
      user_id:     user_id,
      query:       query,
      clicked_url: url,
      rank:        rank,
      searched_at: Time.current
    )
  end
 
  def self.current_affinity(user_id, domain)
    Rails.cache.read("affinity:#{user_id}:#{domain}") || 0.0
  end
 
  # ユーザーの検索履歴からクエリサジェストを生成
  def self.query_suggestions(user_id, prefix)
    recent_queries = SearchLog
      .where(user_id: user_id)
      .where(searched_at: HISTORY_WINDOW.ago..)
      .where("query LIKE ?", "#{prefix}%")
      .group(:query)
      .order(Arel.sql('COUNT(*) DESC'))
      .limit(5)
      .pluck(:query)
 
    # 個人履歴 + 全体の人気クエリをマージ
    popular = PopularQuery.suggestions_for(prefix, limit: 5)
    (recent_queries + popular).uniq.first(8)
  end
end

// spellcheck/corrector.go
package spellcheck
 
// Levenshtein距離（編集距離）の計算
// 2つの文字列の最小編集操作数（挿入・削除・置換）
func LevenshteinDistance(s, t string) int {
	sr := []rune(s)
	tr := []rune(t)
	m, n := len(sr), len(tr)
 
	dp := make([][]int, m+1)
	for i := range dp {
		dp[i] = make([]int, n+1)
	}
	for i := 0; i <= m; i++ { dp[i][0] = i }
	for j := 0; j <= n; j++ { dp[0][j] = j }
 
	for i := 1; i <= m; i++ {
		for j := 1; j <= n; j++ {
			cost := 1
			if sr[i-1] == tr[j-1] {
				cost = 0
			}
			dp[i][j] = min3(
				dp[i-1][j]+1,
				dp[i][j-1]+1,
				dp[i-1][j-1]+cost,
			)
		}
	}
	return dp[m][n]
}
 
func min3(a, b, c int) int {
	if b < a { a = b }
	if c < a { a = c }
	return a
}
 
// BKツリーで編集距離2以内の候補を高速検索
type SpellCorrector struct {
	bkTree *BKTree
}
 
func (sc *SpellCorrector) Correct(word string) string {
	candidates := sc.bkTree.Search(word, 2)
	if len(candidates) == 0 {
		return word
	}
	// 出現頻度が最も高い候補を選択
	best := candidates[0]
	bestFreq := WordFrequency(best)
	for _, c := range candidates[1:] {
		if f := WordFrequency(c); f > bestFreq {
			best, bestFreq = c, f
		}
	}
	return best
}

# app/services/search_monitor.rb
class SearchMonitor
  NAMESPACE = 'SearchEngine/Metrics'
 
  def self.record_search(query:, latency_ms:, result_count:, cache_hit:)
    cw = Aws::CloudWatch::Client.new
    cw.put_metric_data(
      namespace: NAMESPACE,
      metric_data: [
        { metric_name: 'SearchLatency',  value: latency_ms,              unit: 'Milliseconds' },
        { metric_name: 'ResultCount',    value: result_count,            unit: 'Count' },
        { metric_name: 'CacheHitRate',   value: cache_hit ? 1 : 0,       unit: 'Count' },
        { metric_name: 'ZeroResultRate', value: result_count.zero? ? 1 : 0, unit: 'Count' }
      ]
    )
  end
 
  def self.record_crawl(url:, success:, status_code:, latency_ms:)
    cw = Aws::CloudWatch::Client.new
    cw.put_metric_data(
      namespace: NAMESPACE,
      metric_data: [
        { metric_name: 'CrawlSuccess',  value: success ? 1 : 0,                     unit: 'Count' },
        { metric_name: 'CrawlLatency',  value: latency_ms,                          unit: 'Milliseconds' },
        { metric_name: 'HTTP4xxErrors', value: (400..499).cover?(status_code) ? 1 : 0, unit: 'Count' }
      ]
    )
  end
end

# CloudWatch Alarms（主要なSLO監視）
 
SearchLatencyP99:
  メトリクス: SearchLatency（p99）
  閾値: 500ms超過
  評価期間: 5分
  アクション: SNS → PagerDuty
 
IndexingDelay:
  メトリクス: クロール完了からOpenSearch反映までの遅延
  閾値: 3時間超過
  評価期間: 30分
 
CrawlSuccessRate:
  メトリクス: CrawlSuccess（成功率）
  閾値: 95%未満
  評価期間: 15分
 
ZeroResultRate:
  メトリクス: ZeroResultRate（ゼロ件率）
  閾値: 10%超過
  評価期間: 10分
  用途: インデックス消失・シャード障害の早期検知
 
SLO目標:
  検索レイテンシ p99: 500ms以内
  クロール成功率: 95%以上
  インデックス反映遅延: 3時間以内
  可用性: 99.99%（月間ダウンタイム52分以内）