// ============================================================================= // Meilisearch 业务操作封装 // 提供CRUD操作方法,支持多数据源 // ============================================================================= package meilisearch import ( "context" "fmt" "time" "gitea.com/red-future/common/utils" "github.com/gogf/gf/v2/container/gvar" "github.com/gogf/gf/v2/errors/gerror" "github.com/gogf/gf/v2/frame/g" "github.com/gogf/gf/v2/os/glog" "github.com/gogf/gf/v2/os/gtime" "github.com/gogf/gf/v2/util/gconv" ms "github.com/meilisearch/meilisearch-go" ) // ============================================================================= // 向后兼容的Meilisearch结构体 // ============================================================================= type meilisearchDB struct { noCache bool dataSource string // 数据源名称,默认为 "default" } func DB(cache ...bool) *meilisearchDB { return &meilisearchDB{ noCache: false, dataSource: "default", } } // WithDataSource 指定使用的数据源 func (m *meilisearchDB) WithDataSource(name string) *meilisearchDB { m.dataSource = name return m } // NoCache 不使用缓存 func (m *meilisearchDB) NoCache() *meilisearchDB { m.noCache = true return m } // ============================================================================= // 全局变量 // ============================================================================= var ( manager = GetManager() ) const PageSize = 20 // ============================================================================= // Meilisearch 操作方法(支持多数据源) // ============================================================================= // getDataSource 获取当前使用的数据源 func (m *meilisearchDB) getDataSource() (DataSource, error) { if m.dataSource == "" { m.dataSource = "default" } return manager.GetDataSource(m.dataSource) } // getClient 获取 Meilisearch 客户端 func (m *meilisearchDB) getClient() (interface{ Index(string) interface{} }, error) { source, err := m.getDataSource() if err != nil { return nil, err } if c, ok := source.Client().(interface{ Index(string) interface{} }); ok { return c, nil } return nil, fmt.Errorf("invalid client type") } // indexInterface 辅助函数,获取index func indexInterface(indexName string, client interface{ Index(string) interface{} }) interface{} { return client.Index(indexName) } // buildSearchRequest 构建搜索请求 func (m *meilisearchDB) buildSearchRequest(ctx context.Context, searchParams *SearchParams) (*ms.SearchRequest, error) { user, err := utils.GetUserInfo(ctx) if err != nil { return nil, err } req := &ms.SearchRequest{ Limit: int64(PageSize), Page: int64(0), } // 设置查询 if searchParams.Query != "" { req.Query = searchParams.Query } // 设置分页 if searchParams.Page > 0 { req.Page = int64(searchParams.Page - 1) } if searchParams.Limit > 0 { req.Limit = int64(searchParams.Limit) } // 设置排序 if len(searchParams.Sort) > 0 { req.Sort = searchParams.Sort } // 设置过滤条件(包含租户过滤和软删除过滤) filter := "" if !g.IsEmpty(user.TenantId) { filter = fmt.Sprintf("tenantId = %s", gconv.String(user.TenantId)) } if filter == "" { filter = "isDeleted = false" } else { filter += " AND isDeleted = false" } // 添加用户自定义过滤条件 if searchParams.Filter != "" { if filter == "" { filter = searchParams.Filter } else { filter += " AND " + searchParams.Filter } } if filter != "" { req.Filter = filter } // 设置可搜索字段 if searchParams.SearchableAttributes != "" { req.AttributesToSearchOn = []string{searchParams.SearchableAttributes} } // 设置返回字段 if len(searchParams.AttributesToRetrieve) > 0 { req.AttributesToRetrieve = searchParams.AttributesToRetrieve } return req, nil } // Search 搜索文档 func (m *meilisearchDB) Search(ctx context.Context, searchParams *SearchParams, indexName string, result interface{}) (total int64, err error) { client, err := m.getClient() if err != nil { return 0, err } // 构建搜索请求 req, err := m.buildSearchRequest(ctx, searchParams) if err != nil { return 0, err } // Redis 缓存处理 user, err := utils.GetUserInfo(ctx) if err != nil { return } cacheKey := fmt.Sprintf("meilisearch:search:%s:%s:%+v", user.TenantId, indexName, searchParams) if !m.noCache { var resultStr *gvar.Var resultStr, err = g.Redis().Get(ctx, cacheKey) if err != nil { return } if !g.IsEmpty(resultStr) { searchResult := &SearchResult{} if err = gconv.Struct(resultStr, searchResult); err != nil { return } total = int64(searchResult.EstimatedTotalHits) if len(searchResult.Hits) > 0 { if resultArr, ok := result.(*[]map[string]interface{}); ok { *resultArr = searchResult.Hits } else { err = gconv.Structs(searchResult.Hits, result) if err != nil { return } } } return } } // 执行搜索 idx := indexInterface(indexName, client) var searchResp *ms.SearchResponse if i, ok := idx.(interface { Search(string, *ms.SearchRequest) (*ms.SearchResponse, error) }); ok { searchResp, err = i.Search(searchParams.Query, req) } else { return 0, fmt.Errorf("index does not support Search method") } if err != nil { return 0, err } total = int64(searchResp.EstimatedTotalHits) // 解析结果 if len(searchResp.Hits) > 0 { hits := make([]map[string]interface{}, 0, len(searchResp.Hits)) for _, hit := range searchResp.Hits { hitMap := gconv.Map(hit) // 移除 Meilisearch 内部字段 delete(hitMap, "_formatted") hits = append(hits, hitMap) } if resultArr, ok := result.(*[]map[string]interface{}); ok { *resultArr = hits } else { err = gconv.Structs(hits, result) if err != nil { return } } } // 写入缓存 if !m.noCache { hitList := make([]map[string]interface{}, 0) if len(searchResp.Hits) > 0 { for _, hit := range searchResp.Hits { hitMap := gconv.Map(hit) delete(hitMap, "_formatted") hitList = append(hitList, hitMap) } } searchResult := &SearchResult{ Hits: hitList, EstimatedTotalHits: searchResp.EstimatedTotalHits, Limit: int(searchResp.Limit), Offset: int(searchResp.Offset), ProcessingTimeMs: int(searchResp.ProcessingTimeMs), } err = g.Redis().SetEX(ctx, cacheKey, searchResult, int64(time.Hour)) if err != nil { return } } return } // Insert 插入文档 func (m *meilisearchDB) Insert(ctx context.Context, document interface{}, indexName string) (taskUID int64, err error) { c, err := m.getClient() if err != nil { return 0, err } user, err := utils.GetUserInfo(ctx) if err != nil { return } // 转换为 map docMap := gconv.Map(document) // 设置租户ID if !g.IsEmpty(user.TenantId) && g.IsEmpty(docMap["tenantId"]) { docMap["tenantId"] = user.TenantId } // 设置创建人 if !g.IsEmpty(user.UserName) && g.IsEmpty(docMap["creator"]) { docMap["creator"] = user.UserName } // 设置更新人 if !g.IsEmpty(user.UserName) && g.IsEmpty(docMap["updater"]) { docMap["updater"] = user.UserName } // 设置时间 now := gtime.Now().Time if g.IsEmpty(docMap["createdAt"]) { docMap["createdAt"] = now.Unix() } if g.IsEmpty(docMap["updatedAt"]) { docMap["updatedAt"] = now.Unix() } // 设置删除标记 if g.IsEmpty(docMap["isDeleted"]) { docMap["isDeleted"] = false } // 执行插入 documents := []map[string]interface{}{docMap} idx := indexInterface(indexName, c) var task *ms.TaskInfo if i, ok := idx.(interface { AddDocuments([]map[string]interface{}, interface{}) (*ms.TaskInfo, error) }); ok { task, err = i.AddDocuments(documents, nil) } else { return 0, fmt.Errorf("index does not support AddDocuments method") } if err != nil { return 0, err } // 清理缓存 err = m.cleanCache(ctx, indexName, user.TenantId) if err != nil { glog.Warning(ctx, "清理Redis缓存失败:", err) } return task.TaskUID, nil } // InsertMany 批量插入文档 func (m *meilisearchDB) InsertMany(ctx context.Context, documents []interface{}, indexName string) (taskUID int64, err error) { c, err := m.getClient() if err != nil { return 0, err } user, err := utils.GetUserInfo(ctx) if err != nil { return 0, err } docs := make([]map[string]interface{}, 0, len(documents)) for _, document := range documents { docMap := gconv.Map(document) // 设置租户ID if !g.IsEmpty(user.TenantId) && g.IsEmpty(docMap["tenantId"]) { docMap["tenantId"] = user.TenantId } // 设置创建人 if !g.IsEmpty(user.UserName) && g.IsEmpty(docMap["creator"]) { docMap["creator"] = user.UserName } // 设置更新人 if !g.IsEmpty(user.UserName) && g.IsEmpty(docMap["updater"]) { docMap["updater"] = user.UserName } // 设置时间 now := gtime.Now().Time if g.IsEmpty(docMap["createdAt"]) { docMap["createdAt"] = now.Unix() } if g.IsEmpty(docMap["updatedAt"]) { docMap["updatedAt"] = now.Unix() } // 设置删除标记 if g.IsEmpty(docMap["isDeleted"]) { docMap["isDeleted"] = false } docs = append(docs, docMap) } // 执行批量插入 idx := indexInterface(indexName, c) var task *ms.TaskInfo if i, ok := idx.(interface { AddDocuments([]map[string]interface{}, interface{}) (*ms.TaskInfo, error) }); ok { task, err = i.AddDocuments(docs, nil) } else { return 0, fmt.Errorf("index does not support AddDocuments method") } if err != nil { return 0, err } // 清理缓存 err = m.cleanCache(ctx, indexName, user.TenantId) if err != nil { glog.Warning(ctx, "清理Redis缓存失败:", err) } return task.TaskUID, nil } // Update 更新文档 func (m *meilisearchDB) Update(ctx context.Context, document interface{}, indexName string) (taskUID int64, err error) { c, err := m.getClient() if err != nil { return 0, err } user, err := utils.GetUserInfo(ctx) if err != nil { return 0, err } // 转换为 map docMap := gconv.Map(document) // 设置更新人 if !g.IsEmpty(user.UserName) && g.IsEmpty(docMap["updater"]) { docMap["updater"] = user.UserName } // 设置更新时间 docMap["updatedAt"] = gtime.Now().Unix() // 执行更新 documents := []map[string]interface{}{docMap} idx := indexInterface(indexName, c) var task *ms.TaskInfo if i, ok := idx.(interface { UpdateDocuments([]map[string]interface{}, interface{}) (*ms.TaskInfo, error) }); ok { task, err = i.UpdateDocuments(documents, nil) } else { return 0, fmt.Errorf("index does not support UpdateDocuments method") } if err != nil { return 0, err } // 清理缓存 err = m.cleanCache(ctx, indexName, user.TenantId) if err != nil { glog.Warning(ctx, "清理Redis缓存失败:", err) } return task.TaskUID, nil } // Delete 删除文档 func (m *meilisearchDB) Delete(ctx context.Context, id string, indexName string) (taskUID int64, err error) { c, err := m.getClient() if err != nil { return 0, err } // 执行删除 idx := indexInterface(indexName, c) var task *ms.TaskInfo if i, ok := idx.(interface { DeleteDocument(string) (*ms.TaskInfo, error) }); ok { task, err = i.DeleteDocument(id) } else { return 0, fmt.Errorf("index does not support DeleteDocument method") } if err != nil { return 0, err } // 清理缓存 user, err := utils.GetUserInfo(ctx) if err != nil { return } err = m.cleanCache(ctx, indexName, user.TenantId) if err != nil { glog.Warning(ctx, "清理Redis缓存失败:", err) } return task.TaskUID, nil } // DeleteSoft 软删除文档 func (m *meilisearchDB) DeleteSoft(ctx context.Context, id string, indexName string) (taskUID int64, err error) { c, err := m.getClient() if err != nil { return 0, err } user, err := utils.GetUserInfo(ctx) if err != nil { return 0, err } // 软删除:更新 isDeleted 字段 updateMap := map[string]interface{}{ "id": id, "isDeleted": true, "updater": user.UserName, "updatedAt": gtime.Now().Unix(), } // 执行更新 documents := []map[string]interface{}{updateMap} idx := indexInterface(indexName, c) var task *ms.TaskInfo if i, ok := idx.(interface { UpdateDocuments([]map[string]interface{}, interface{}) (*ms.TaskInfo, error) }); ok { task, err = i.UpdateDocuments(documents, nil) } else { return 0, fmt.Errorf("index does not support UpdateDocuments method") } if err != nil { return 0, err } // 清理缓存 err = m.cleanCache(ctx, indexName, user.TenantId) if err != nil { glog.Warning(ctx, "清理Redis缓存失败:", err) } return task.TaskUID, nil } // Get 获取单个文档 func (m *meilisearchDB) Get(ctx context.Context, id string, indexName string, result interface{}) (err error) { c, err := m.getClient() if err != nil { return err } // Redis 缓存处理 user, err := utils.GetUserInfo(ctx) if err != nil { return } cacheKey := fmt.Sprintf("meilisearch:doc:%s:%s:%s", user.TenantId, indexName, id) if !m.noCache { var resultStr *gvar.Var resultStr, err = g.Redis().Get(ctx, cacheKey) if err != nil { return } if !g.IsEmpty(resultStr) { return gconv.Scan(resultStr, result) } } // 执行查询 var doc map[string]interface{} idx := indexInterface(indexName, c) if i, ok := idx.(interface { GetDocument(string, interface{}) error }); ok { err = i.GetDocument(id, &doc) } else { return fmt.Errorf("index does not support GetDocument method") } if err != nil { return err } // 过滤已删除的文档 if gconv.Bool(doc["isDeleted"]) { return gerror.New("文档不存在") } err = gconv.Struct(doc, result) if err != nil { return err } // 写入缓存 if !m.noCache { err = g.Redis().SetEX(ctx, cacheKey, result, int64(time.Hour)) if err != nil { return err } } return nil } // cleanCache 清理缓存 func (m *meilisearchDB) cleanCache(ctx context.Context, indexName string, tenantId interface{}) error { // 清理搜索缓存 searchKeys, err := g.Redis().Keys(ctx, fmt.Sprintf("meilisearch:search:%s:%s:*", tenantId, indexName)) if err != nil { return err } for _, key := range searchKeys { _, err = g.Redis().Del(ctx, key) if err != nil { return err } } return nil } // CreateIndex 创建索引 func (m *meilisearchDB) CreateIndex(ctx context.Context, indexConfig *IndexConfig) (taskUID int64, err error) { client, err := m.getClient() if err != nil { return 0, err } indexSettings := &ms.IndexConfig{ Uid: indexConfig.UID, PrimaryKey: indexConfig.PrimaryKey, } if c, ok := client.(interface { CreateIndex(*ms.IndexConfig) (*ms.TaskInfo, error) }); ok { task, err := c.CreateIndex(indexSettings) if err != nil { return 0, err } return task.TaskUID, nil } return 0, fmt.Errorf("client does not support CreateIndex") } // DeleteIndex 删除索引 func (m *meilisearchDB) DeleteIndex(ctx context.Context, indexName string) (err error) { client, err := m.getClient() if err != nil { return err } if c, ok := client.(interface{ DeleteIndex(string) error }); ok { return c.DeleteIndex(indexName) } return fmt.Errorf("client does not support DeleteIndex") } // GetIndex 获取索引信息 func (m *meilisearchDB) GetIndex(ctx context.Context, indexName string) (interface{}, error) { client, err := m.getClient() if err != nil { return nil, err } if c, ok := client.(interface { GetIndex(string) (interface{}, error) }); ok { return c.GetIndex(indexName) } return nil, fmt.Errorf("client does not support GetIndex") } // GetIndexes 获取所有索引 func (m *meilisearchDB) GetIndexes(ctx context.Context) (interface{}, error) { client, err := m.getClient() if err != nil { return nil, err } if c, ok := client.(interface { GetIndexes(interface{}) (interface{}, error) }); ok { return c.GetIndexes(nil) } return nil, fmt.Errorf("client does not support GetIndexes") } // UpdateSettings 更新索引设置 func (m *meilisearchDB) UpdateSettings(ctx context.Context, indexName string, settings *ms.Settings) (taskUID int64, err error) { c, err := m.getClient() if err != nil { return 0, err } idx := indexInterface(indexName, c) var task *ms.TaskInfo if i, ok := idx.(interface { UpdateSettings(*ms.Settings) (*ms.TaskInfo, error) }); ok { task, err = i.UpdateSettings(settings) } else { return 0, fmt.Errorf("index does not support UpdateSettings method") } if err != nil { return 0, err } return task.TaskUID, nil } // GetSettings 获取索引设置 func (m *meilisearchDB) GetSettings(ctx context.Context, indexName string) (*ms.Settings, error) { c, err := m.getClient() if err != nil { return nil, err } idx := indexInterface(indexName, c) var settings *ms.Settings if i, ok := idx.(interface{ GetSettings() (*ms.Settings, error) }); ok { settings, err = i.GetSettings() } else { return nil, fmt.Errorf("index does not support GetSettings method") } if err != nil { return nil, err } return settings, nil } // GetClient 获取原始客户端(用于高级操作) func (m *meilisearchDB) GetClient() (interface{ Index(string) interface{} }, error) { return m.getClient() } // BuildUpdateData 构建更新数据 func BuildUpdateData(ctx context.Context, req interface{}) (map[string]interface{}, error) { return gconv.Map(req), nil }