ElasticSearch 与搜索服务

# docker-compose.yml 简化配置
services:
  elasticsearch:
    image: docker.elastic.co/elasticsearch/elasticsearch:8.0.0
    environment:
      - discovery.type=single-node    # 单节点模式
      - xpack.security.enabled=false  # 禁用 xpack（生产不要禁用）
    ports:
      - "9200:9200"

PUT /user_idx
{
  "settings": {
    "number_of_shards": 3,    # 3 个分片
    "number_of_replicas": 2   # 2 个副本
  },
  "mappings": {
    "properties": {
      "nickname": { "type": "text" },
      "email":    { "type": "keyword" },
      "phone":    { "type": "keyword" }
    }
  }
}

POST /user_idx/_doc
{
  "nickname": "大明",
  "email": "john@example.com",
  "phone": "13800000000"
}

POST /user_idx/_search
{
  "query": {
    "match": { "email": "example.com" }
  }
}

package es

import (
    "context"
    "fmt"
    "github.com/elastic/go-elasticsearch/v8"
)

// NewClient 初始化 ES 客户端
func NewClient(addresses []string) (*elasticsearch.Client, error) {
    // 步骤 1：构造客户端配置（可配置用户名密码、TLS 等）
    cfg := elasticsearch.Config{
        Addresses: addresses,
    }
    // 步骤 2：创建客户端
    es, err := elasticsearch.NewClient(cfg)
    if err != nil {
        return nil, fmt.Errorf("创建 ES 客户端失败: %w", err)
    }
    // 步骤 3：健康检查，确认集群可达
    resp, err := es.Info()
    if err != nil {
        return nil, err
    }
    defer resp.Body.Close()
    fmt.Println("ES 连接成功", resp)
    return es, nil
}

// User 索引定义
// 隐私字段（email、phone）的处理因产品形态而异：
// - toC 公开平台：不允许通过邮箱 / 手机号搜索
// - toB 私有 IM：需要支持邮箱 / 手机号搜索
type User struct {
    ID       int64  `json:"id"`
    Nickname string `json:"nickname"` // 通常设为 text + keyword 双类型
    Email    string `json:"email"`    // keyword 类型
    Phone    string `json:"phone"`    // keyword 类型
}

// SaveUser 保存用户数据到 ES
// 关键：指定文档 ID = 用户 ID，实现 Insert Or Update 语义
// 这样用户注册、更新信息时，同一用户始终对应同一份文档
func (r *UserESRepository) SaveUser(ctx context.Context, user domain.User) error {
    // 步骤 1：把领域对象映射成 ES 文档结构
    doc := User{
        ID:       user.ID,
        Nickname: user.Nickname,
        Email:    user.Email,
        Phone:    user.Phone,
    }
    // 步骤 2：序列化成 JSON body
    body, _ := json.Marshal(doc)
    // 步骤 3：用文档 ID 作为 ES 文档的 _id，达到 Upsert 效果
    _, err := r.es.Index().
        Index("user_idx").
        Id(fmt.Sprintf("%d", user.ID)).  // 关键：指定 _id
        Body(string(body)).
        Do(ctx)
    return err
}

// Article 帖子索引定义
// 注意：通常不存储作者信息（如作者名），因为搜索人名应该走用户索引
// 保留 status 字段：文章设为不可见时，不应被搜索到
type Article struct {
    ID      int64  `json:"id"`
    Title   string `json:"title"`    // text 类型，全文搜索
    Content string `json:"content"`  // text 类型
    Status  uint8  `json:"status"`   // keyword 类型，过滤已发表状态
}

// SearchService 在 Service 层聚合多业务搜索结果
// 大型平台的完整流程：
// 1. 输入预处理（去空格、分词、同义词替换）
// 2. 解析表达式生成搜索计划
// 3. 执行搜索计划
// 4. 根据业务规则调整最终输出
// 本课程实现简化版：分别查询 User 和 Article，聚合返回
func (s *SearchService) Search(ctx context.Context, uid int64, expression string) (SearchResult, error) {
    // 步骤 1：输入预处理：trim、切割关键字
    keywords := strings.Fields(strings.TrimSpace(expression))
    // 步骤 2：并发查询多个业务方（可使用 errgroup 加速）
    var (
        wg    sync.WaitGroup
        users []domain.User
        arts  []domain.Article
    )
    wg.Add(2)
    go func() {
        defer wg.Done()
        users, _ = s.userRepo.Search(ctx, keywords)
    }()
    go func() {
        defer wg.Done()
        arts, _ = s.articleRepo.Search(ctx, keywords)
    }()
    wg.Wait()
    // 步骤 3：聚合返回
    return SearchResult{Users: users, Articles: arts}, nil
}

// SearchArticle 利用 ES 多关键字匹配特性
// 必须过滤 status = 已发表
func (r *ArticleESRepository) SearchArticle(ctx context.Context, keywords []string) ([]domain.Article, error) {
    // 步骤 1：构造 bool 查询：must 多字段匹配 + filter 状态过滤
    query := map[string]interface{}{
        "query": map[string]interface{}{
            "bool": map[string]interface{}{
                "must": []interface{}{
                    map[string]interface{}{
                        "multi_match": map[string]interface{}{
                            "query":  strings.Join(keywords, " "),
                            "fields": []string{"title", "content"},
                        },
                    },
                },
                "filter": []interface{}{
                    map[string]interface{}{
                        "term": map[string]interface{}{"status": uint8(2)}, // 已发表
                    },
                },
            },
        },
    }
    // 步骤 2：序列化并发送给 ES
    body, _ := json.Marshal(query)
    resp, err := r.es.Search().
        Index("article_idx").
        Body(string(body)).
        Do(ctx)
    if err != nil {
        return nil, err
    }
    // 步骤 3：解析返回结果集...
    return parseArticles(resp), nil
}

// Tag 标签表
type Tag struct {
    ID   int64  `gorm:"primaryKey;autoIncrement"`
    UID  int64  `gorm:"index:idx_uid_name,unique,priority:1;comment:创建者"`
    Name string `gorm:"type:varchar(128);index:idx_uid_name,unique,priority:2;comment:标签名"`
    Ctime int64
    Utime int64
}
// 唯一索引 (uid, name) 防止同一用户创建重名标签

// TagBiz 资源-标签关联表
type TagBiz struct {
    ID      int64  `gorm:"primaryKey;autoIncrement"`
    UID     int64  `gorm:"index:idx_uid_biz,priority:1;comment:打标签的人"`
    Biz     string `gorm:"type:varchar(128);index:idx_uid_biz,priority:2"`
    BizID   int64  `gorm:"index:idx_uid_biz,priority:3"`
    TID     int64  `gorm:"index:idx_tid;comment:标签ID"`
    Ctime   int64
    Utime   int64
}

// CreateTag 创建标签
// 标签一旦创建即不可变，所以可以直接放入缓存
func (s *TagService) CreateTag(ctx context.Context, uid int64, name string) (int64, error) {
    // 步骤 1：落库创建标签
    id, err := s.repo.CreateTag(ctx, domain.Tag{UID: uid, Name: name})
    if err != nil {
        return 0, err
    }
    // 步骤 2：假设单用户无并发操作，直接将新标签加入缓存
    _ = s.cache.Append(ctx, uid, name)
    return id, nil
}

// Append 将标签追加到用户标签列表
// 前提：标签不可变（无更新、无删除），所以可以放心缓存
// 使用 pipeline 提高性能；也可用 lua 或 TxPipeline
func (c *TagCache) Append(ctx context.Context, uid int64, name string) error {
    return c.client.RPush(ctx, tagKey(uid), name).Err()
}

// PreloadUserTags 启动时批量加载所有用户标签到缓存
// 适用于 toB 场景：所有用户标签总数不大，可全部缓存
func (r *TagRepository) PreloadUserTags(ctx context.Context) error {
    // 步骤 1：分批游标，从 UID=0 开始往后扫
    const batchSize = 100
    var lastUID int64 = 0
    for {
        // 步骤 2：按批从 DB 拉取
        tags, err := r.findBatchByUID(ctx, lastUID, batchSize)
        if err != nil {
            return err
        }
        if len(tags) == 0 {
            break
        }
        // 步骤 3：按 UID 分组，pipeline 写入 Redis
        groups := groupByUID(tags)
        for uid, names := range groups {
            _ = r.cache.BatchAppend(ctx, uid, names)
        }
        lastUID = tags[len(tags)-1].UID
    }
    return nil
}

// 在应用启动时调用
func main() {
    // ... 初始化 Repository
    repo := tag.NewTagRepository(...)
    // 启动后预加载缓存
    go func() {
        if err := repo.PreloadUserTags(context.Background()); err != nil {
            log.Warn("预加载标签缓存失败", logger.Error(err))
        }
    }()
}

// AttachTags 给资源打标签，覆盖原有标签
func (s *TagService) AttachTags(ctx context.Context, uid int64, biz string, bizID int64, tids []int64) error {
    return s.repo.AttachTags(ctx, uid, biz, bizID, tids)
}

// 实现思路：
// 1. 先删除该 (uid, biz, biz_id) 下的所有关联
// 2. 再插入新的关联
func (r *TagRepository) AttachTags(ctx context.Context, uid int64, biz string, bizID int64, tids []int64) error {
    // 步骤 1：开事务，先删除旧关联
    return r.db.WithContext(ctx).Transaction(func(tx *gorm.DB) error {
        // 1. 删除旧关联
        if err := tx.Where("uid = ? AND biz = ? AND biz_id = ?", uid, biz, bizID).
            Delete(&TagBiz{}).Error; err != nil {
            return err
        }
        // 2. 插入新关联
        if len(tids) == 0 {
            return nil
        }
        rows := make([]TagBiz, 0, len(tids))
        for _, tid := range tids {
            rows = append(rows, TagBiz{UID: uid, Biz: biz, BizID: bizID, TID: tid})
        }
        return tx.Create(&rows).Error
    })
}

// GetTags 查询资源上的标签
// 模型始终是"某人给某资源打了某标签"，所以查询要带 uid
// 支持系统默认标签：用 uid = -1 表示
//   合并 uid = 123（真实用户）和 uid = -1（系统）的结果
func (r *TagRepository) GetTags(ctx context.Context, uid int64, biz string, bizID int64) ([]domain.Tag, error) {
    // 步骤 1：查关联表，并用 Preload 一次 JOIN 出标签名
    var res []TagBiz
    err := r.db.WithContext(ctx).
        Preload("Tag").  // 通过 Preload 一次 JOIN 出标签名
        Where("uid IN ? AND biz = ? AND biz_id = ?", []int64{uid, -1}, biz, bizID).
        Find(&res).Error
    if err != nil {
        return nil, err
    }
    // 步骤 2：转换为 domain.Tag
    tags := make([]domain.Tag, 0, len(res))
    for _, tb := range res {
        tags = append(tags, domain.Tag{ID: tb.Tag.ID, Name: tb.Tag.Name})
    }
    return tags, nil
}

// Kafka 消息定义（与 sync.proto 一致）
type SyncDataEvent struct {
    IndexName string                 `json:"index_name"`  // 索引名
    Data      map[string]interface{} `json:"data"`        // 文档数据
}

// 推送标签
func (s *TagService) SyncTagToSearch(ctx context.Context, tag domain.Tag) error {
    // 步骤 1：组装统一格式事件（index_name + data），与搜索服务约定
    event := SyncDataEvent{
        IndexName: "tag_idx",
        Data: map[string]interface{}{
            "id":   tag.ID,
            "uid":  tag.UID,
            "name": tag.Name,
        },
    }
    // 步骤 2：序列化并发送；用 hash(uid) 作 key 保证同用户标签进同一分区，保序
    msg, _ := json.Marshal(event)
    // 同步推送（标签打标不是高并发场景，Kafka 性能足够）
    // 注意：用 hash(uid) 作为 key，保证同一用户的标签消息发到同一 partition，保持顺序
    return s.producer.SendMessage(ctx, "sync_data", fmt.Sprintf("%d", tag.UID), msg)
}

// tag_idx 索引
// 标签始终与个人挂钩，不适合放进 Article 或 User 索引
type TagDoc struct {
    ID   int64  `json:"id"`
    UID  int64  `json:"uid"`
    Name string `json:"name"`  // text 类型，支持模糊搜索
}

// 步骤 1：先查标签索引，找到匹配的 biz_id
// 步骤 2：再查 article 索引，叠加是否命中标签的条件，并赋予高权重（boost）
func (r *SearchRepository) SearchArticleWithTagBoost(ctx context.Context, keywords []string) ([]domain.Article, error) {
    // 步骤 1：查询标签索引，拿到命中关键字的 biz_id 列表
    tagBizIDs, err := r.findBizIDsByTag(ctx, keywords)
    if err != nil {
        return nil, err
    }
    // 步骤 2：查询文章索引，用 bool.should 叠加基础匹配 + 标签命中 boost
    query := map[string]interface{}{
        "query": map[string]interface{}{
            "bool": map[string]interface{}{
                "should": []interface{}{
                    // 基础查询：title + content 匹配
                    map[string]interface{}{
                        "multi_match": map[string]interface{}{
                            "query":  strings.Join(keywords, " "),
                            "fields": []string{"title", "content"},
                        },
                    },
                    // 增强查询：命中标签的文章，boost 提升权重
                    map[string]interface{}{
                        "terms": map[string]interface{}{
                            "id":    tagBizIDs,
                            "boost": 2.0,  // boost 越高越相关，默认 1.0
                        },
                    },
                },
            },
        },
    }
    body, _ := json.Marshal(query)
    resp, err := r.es.Search().Index("article_idx").Body(string(body)).Do(ctx)
    // 解析返回...
    return parseArticles(resp), nil
}