微型博客全文搜索：从 PostgreSQL 全文检索到 Elasticsearch

-- 创建短文表的全文检索列
ALTER TABLE posts ADD COLUMN content_tsv tsvector;

-- 利用 GENERATED ALWAYS AS 自动维护
ALTER TABLE posts
  ADD COLUMN content_tsv tsvector
  GENERATED ALWAYS AS (to_tsvector('simple', content)) STORED;

-- 建立 GIN 索引
CREATE INDEX idx_posts_content_tsv ON posts USING GIN (content_tsv);

-- 检索：匹配包含「架构」与「分布式」的短文
SELECT id, content
FROM posts
WHERE content_tsv @@ to_tsquery('simple', '架构 & 分布式')
ORDER BY ts_rank(content_tsv, to_tsquery('simple', '架构 & 分布式')) DESC
LIMIT 20;

-- PostgreSQL 内置 simple 配置对中文的效果
SELECT to_tsvector('simple', '探索分布式架构设计');
-- 结果：'探索分布式架构设计'（整串作为一个词元，几乎无法切分）

-- 方案 2 示意：应用层预分词 + 数组 GIN 索引
CREATE TABLE post_search_terms (
    post_id BIGINT PRIMARY KEY REFERENCES posts(id),
    terms TEXT[] NOT NULL   -- ['分布式', '架构', '设计', '探索']
);
CREATE INDEX idx_post_search_terms ON post_search_terms USING GIN (terms);

SELECT p.id, p.content
FROM posts p
JOIN post_search_terms t ON t.post_id = p.id
WHERE t.terms && ARRAY['架构']
ORDER BY p.created_at DESC
LIMIT 20;

{
  "settings": {
    "number_of_shards": 3,
    "number_of_replicas": 1,
    "analysis": {
      "analyzer": "ik_max_word",
      "filter": {
        "pinyin_filter": { "type": "pinyin", "keep_full_pinyin": true }
      }
    }
  },
  "mappings": {
    "properties": {
      "content": { "type": "text", "analyzer": "ik_max_word" },
      "username": { "type": "keyword" },
      "hashtags": { "type": "keyword" },
      "like_count": { "type": "integer" },
      "repost_count": { "type": "integer" },
      "author_authority": { "type": "float" },
      "created_at": { "type": "date" },
      "visibility": { "type": "keyword" }
    }
  }
}

// 检索时混合使用不同分析器
const query = {
  bool: {
    must: [
      { match: { content: { query: "分布式架构", analyzer: "ik_smart" } } }
    ],
    should: [
      { match: { content: { query: "分布式架构", analyzer: "pinyin_analyzer", boost: 0.2 } } }
    ]
  }
};

// Go 消费端：从 Kafka 接收短文变更事件并写入 ES
package search

import (
	"context"
	"encoding/json"
	"github.com/segmentio/kafka-go"
)

type PostEvent struct {
	Action    string `json:"action"` // create | update | delete
	PostID    int64  `json:"post_id"`
	Content   string `json:"content"`
	Hashtags  []string `json:"hashtags"`
	Username  string `json:"username"`
	LikeCount int64  `json:"like_count"`
	UpdatedAt string `json:"updated_at"`
}

func (s *SearchService) ConsumePostEvents(ctx context.Context, topic string) error {
	reader := kafka.NewReader(kafka.ReaderConfig{
		Brokers: []string{"kafka-1:9092", "kafka-2:9092"},
		Topic:   topic,
		GroupID: "search-indexer",
	})
	defer reader.Close()

	for {
		msg, err := reader.ReadMessage(ctx)
		if err != nil {
			return err
		}
		var ev PostEvent
		if err := json.Unmarshal(msg.Value, &ev); err != nil {
			continue // 跳过脏数据
		}
		switch ev.Action {
		case "delete":
			s.es.Delete("posts", ev.PostID)
		default:
			s.es.Index("posts", ev.PostID, ev)
		}
	}
}

// ES 多路召回查询骨架
const multiRecall = {
  from: 0,
  size: 50,
  query: {
    bool: {
      must: [{ match: { content: query } }],
      should: [
        { match: { hashtags: query, boost: 3 } },
        { term: { username: query, boost: 2 } },
        { match: { content: query, analyzer: "ik_smart", boost: 1 } }
      ],
      filter: [{ term: { visibility: "public" } }]
    }
  }
};

{
  "query": {
    "function_score": {
      "query": { "match": { "content": "分布式架构" } },
      "functions": [
        { "weight": 3, "filter": { "term": { "hashtags": "分布式架构" } } },
        { "weight": 2, "filter": { "range": { "like_count": { "gte": 100 } } } },
        {
          "gauss": {
            "created_at": { "origin": "now", "scale": "7d" }
          }
        }
      ],
      "score_mode": "sum",
      "boost_mode": "multiply"
    }
  }
}

// Go 精排示意：融合社交特征
func rerank(userID int64, hits []Hit) []Hit {
	for i := range hits {
		h := &hits[i]
		// 1. 文本相关度 (ES 已给出)
		textScore := h.BM25
		// 2. 作者权威度 (粉丝数、历史互动归一化)
		authority := authorAuthority(h.AuthorID)
		// 3. 内容质量 (点赞/转发/评论加权)
		quality := qualityScore(h)
		// 4. 时效衰减 (指数衰减)
		freshness := math.Exp(-hoursSince(h.CreatedAt) / 72)
		// 5. 个性化偏好
		interest := userInterest(userID, h.Hashtags)

		h.FinalScore = 0.4*textScore + 0.25*authority +
			0.15*quality + 0.1*freshness + 0.1*interest
	}
	// 稳定排序后截断
	sort.Slice(hits, func(i, j int) bool {
		return hits[i].FinalScore > hits[j].FinalScore
	})
	return hits[:20]
}

// 降级开关示例
func Search(ctx context.Context, q string, userID int64) (*SearchResult, error) {
	if searchDisabled.Load() {
		// 降级到 PostgreSQL 兜底
		return legacySearch(ctx, q)
	}
	res, err := esSearch(ctx, q, userID)
	if err != nil {
		// 单次请求降级，避免雪崩
		if isESUnavailable(err) {
			return legacySearch(ctx, q)
		}
		return nil, err
	}
	return res, nil
}