.gitignore
This commit is contained in:
392
full-text-search/meilisearch/connection.go
Normal file
392
full-text-search/meilisearch/connection.go
Normal file
@@ -0,0 +1,392 @@
|
||||
// =============================================================================
|
||||
// Meilisearch 数据源连接管理
|
||||
// 负责数据源的连接、重连、健康检查和优雅关闭
|
||||
// =============================================================================
|
||||
|
||||
package meilisearch
|
||||
|
||||
import (
|
||||
"context"
|
||||
"fmt"
|
||||
"os"
|
||||
"os/signal"
|
||||
"sync"
|
||||
"syscall"
|
||||
"time"
|
||||
|
||||
"github.com/gogf/gf/v2/frame/g"
|
||||
"github.com/gogf/gf/v2/os/glog"
|
||||
"github.com/gogf/gf/v2/util/gconv"
|
||||
ms "github.com/meilisearch/meilisearch-go"
|
||||
)
|
||||
|
||||
// =============================================================================
|
||||
// 数据源配置结构
|
||||
// =============================================================================
|
||||
|
||||
type DataSourceConfig struct {
|
||||
Name string `json:"name"`
|
||||
Host string `json:"host"`
|
||||
Port int `json:"port"`
|
||||
APIKey string `json:"apiKey"`
|
||||
Timeout time.Duration `json:"timeout"`
|
||||
}
|
||||
|
||||
// =============================================================================
|
||||
// 单个数据源接口
|
||||
// =============================================================================
|
||||
|
||||
type DataSource interface {
|
||||
Name() string
|
||||
Client() interface{}
|
||||
IsConnected() bool
|
||||
Connect(ctx context.Context) error
|
||||
Reconnect(ctx context.Context) error
|
||||
Close(ctx context.Context) error
|
||||
}
|
||||
|
||||
// =============================================================================
|
||||
// 数据源实现
|
||||
// =============================================================================
|
||||
|
||||
type BaseDataSource struct {
|
||||
config *DataSourceConfig
|
||||
client interface{}
|
||||
isConnected bool
|
||||
mu sync.RWMutex
|
||||
lastError error
|
||||
lastErrorTime time.Time
|
||||
}
|
||||
|
||||
func NewBaseDataSource(config *DataSourceConfig) *BaseDataSource {
|
||||
return &BaseDataSource{
|
||||
config: config,
|
||||
isConnected: false,
|
||||
}
|
||||
}
|
||||
|
||||
func (d *BaseDataSource) Name() string {
|
||||
return d.config.Name
|
||||
}
|
||||
|
||||
func (d *BaseDataSource) Client() interface{} {
|
||||
d.mu.RLock()
|
||||
defer d.mu.RUnlock()
|
||||
return d.client
|
||||
}
|
||||
|
||||
func (d *BaseDataSource) IsConnected() bool {
|
||||
d.mu.RLock()
|
||||
defer d.mu.RUnlock()
|
||||
return d.isConnected && d.client != nil
|
||||
}
|
||||
|
||||
func (d *BaseDataSource) Connect(ctx context.Context) error {
|
||||
d.mu.Lock()
|
||||
defer d.mu.Unlock()
|
||||
|
||||
// 构建客户端
|
||||
d.client = ms.New(d.config.Host, ms.WithAPIKey(d.config.APIKey))
|
||||
|
||||
// 测试连接
|
||||
if err := d.healthCheck(ctx); err != nil {
|
||||
d.isConnected = false
|
||||
d.lastError = err
|
||||
d.lastErrorTime = time.Now()
|
||||
return fmt.Errorf("datasource [%s] connection failed: %w", d.config.Name, err)
|
||||
}
|
||||
|
||||
d.isConnected = true
|
||||
d.lastError = nil
|
||||
glog.Infof(ctx, "✅ datasource [%s] connected successfully", d.config.Name)
|
||||
return nil
|
||||
}
|
||||
|
||||
// healthCheck 健康检查
|
||||
func (d *BaseDataSource) healthCheck(ctx context.Context) error {
|
||||
if d.client == nil {
|
||||
return fmt.Errorf("client is nil")
|
||||
}
|
||||
|
||||
// 获取版本信息来测试连接
|
||||
if c, ok := d.client.(interface{ GetVersion() (*ms.Version, error) }); ok {
|
||||
status, err := c.GetVersion()
|
||||
if err != nil {
|
||||
return err
|
||||
}
|
||||
glog.Debugf(ctx, "Meilisearch version: %s", status.PkgVersion)
|
||||
} else {
|
||||
// 如果没有GetVersion方法，尝试其他方法验证连接
|
||||
if _, ok := d.client.(interface {
|
||||
GetIndexes(interface{}) (interface{}, error)
|
||||
}); ok {
|
||||
return nil
|
||||
}
|
||||
}
|
||||
|
||||
return nil
|
||||
}
|
||||
|
||||
func (d *BaseDataSource) Reconnect(ctx context.Context) error {
|
||||
glog.Infof(ctx, "🔄 reconnecting datasource [%s]", d.config.Name)
|
||||
return d.Connect(ctx)
|
||||
}
|
||||
|
||||
func (d *BaseDataSource) Close(ctx context.Context) error {
|
||||
d.mu.Lock()
|
||||
defer d.mu.Unlock()
|
||||
|
||||
// Meilisearch 客户端不需要显式关闭，只重置状态
|
||||
d.client = nil
|
||||
d.isConnected = false
|
||||
glog.Infof(ctx, "datasource [%s] closed", d.config.Name)
|
||||
return nil
|
||||
}
|
||||
|
||||
// =============================================================================
|
||||
// 多数据源管理器
|
||||
// =============================================================================
|
||||
|
||||
type DataSourceManager struct {
|
||||
sources map[string]DataSource
|
||||
mu sync.RWMutex
|
||||
ctx context.Context
|
||||
cancel context.CancelFunc
|
||||
started bool
|
||||
maxRetries int
|
||||
}
|
||||
|
||||
var (
|
||||
globalManager *DataSourceManager
|
||||
managerOnce sync.Once
|
||||
)
|
||||
|
||||
// GetManager 获取全局管理器
|
||||
func GetManager() *DataSourceManager {
|
||||
managerOnce.Do(func() {
|
||||
ctx, cancel := context.WithCancel(context.Background())
|
||||
globalManager = &DataSourceManager{
|
||||
sources: make(map[string]DataSource),
|
||||
ctx: ctx,
|
||||
cancel: cancel,
|
||||
started: false,
|
||||
maxRetries: 3,
|
||||
}
|
||||
})
|
||||
return globalManager
|
||||
}
|
||||
|
||||
// RegisterDataSource 注册数据源
|
||||
func (m *DataSourceManager) RegisterDataSource(config *DataSourceConfig) error {
|
||||
m.mu.Lock()
|
||||
defer m.mu.Unlock()
|
||||
|
||||
if _, exists := m.sources[config.Name]; exists {
|
||||
return fmt.Errorf("datasource [%s] already exists", config.Name)
|
||||
}
|
||||
|
||||
source := NewBaseDataSource(config)
|
||||
m.sources[config.Name] = source
|
||||
return nil
|
||||
}
|
||||
|
||||
// GetDataSource 获取数据源
|
||||
func (m *DataSourceManager) GetDataSource(name string) (DataSource, error) {
|
||||
m.mu.RLock()
|
||||
defer m.mu.RUnlock()
|
||||
|
||||
source, exists := m.sources[name]
|
||||
if !exists {
|
||||
return nil, fmt.Errorf("datasource [%s] not found", name)
|
||||
}
|
||||
return source, nil
|
||||
}
|
||||
|
||||
// GetAllDataSourceNames 获取所有数据源名称
|
||||
func (m *DataSourceManager) GetAllDataSourceNames() []string {
|
||||
m.mu.RLock()
|
||||
defer m.mu.RUnlock()
|
||||
|
||||
names := make([]string, 0, len(m.sources))
|
||||
for name := range m.sources {
|
||||
names = append(names, name)
|
||||
}
|
||||
return names
|
||||
}
|
||||
|
||||
// init 初始化多数据源
|
||||
func init() {
|
||||
ctx := context.Background()
|
||||
|
||||
// 从配置初始化多数据源
|
||||
if err := manager.InitializeFromConfig(ctx); err != nil {
|
||||
glog.Errorf(ctx, "❌ Failed to initialize Meilisearch datasources: %v", err)
|
||||
} else {
|
||||
glog.Infof(ctx, "✅ Meilisearch datasources initialized: %v", manager.GetAllDataSourceNames())
|
||||
}
|
||||
|
||||
// 启动健康检查
|
||||
manager.StartHealthCheck()
|
||||
|
||||
// 设置优雅关闭
|
||||
setupGracefulShutdown()
|
||||
}
|
||||
|
||||
// InitializeFromConfig 从配置初始化数据源
|
||||
// 动态读取 config.yml 中 meilisearch 下的所有配置项
|
||||
func (m *DataSourceManager) InitializeFromConfig(ctx context.Context) error {
|
||||
var firstErr error
|
||||
|
||||
// 获取 meilisearch 配置下的所有子键
|
||||
meilisearchConfig := g.Cfg().MustGet(ctx, "meilisearch")
|
||||
if meilisearchConfig.IsNil() {
|
||||
glog.Warningf(ctx, "no meilisearch configuration found in config.yml")
|
||||
return nil
|
||||
}
|
||||
|
||||
// 将配置转换为 map
|
||||
configMap := meilisearchConfig.Map()
|
||||
if configMap == nil {
|
||||
glog.Warningf(ctx, "meilisearch configuration is not a map")
|
||||
return nil
|
||||
}
|
||||
|
||||
// 遍历所有 meilisearch 子配置
|
||||
for name, subConfig := range configMap {
|
||||
// 跳过非对象类型的配置
|
||||
subMap, ok := subConfig.(map[string]interface{})
|
||||
if !ok {
|
||||
continue
|
||||
}
|
||||
|
||||
// 检查是否有 host 配置
|
||||
host, hasHost := subMap["host"]
|
||||
if !hasHost || gconv.String(host) == "" {
|
||||
continue
|
||||
}
|
||||
|
||||
// 构建数据源配置
|
||||
config := &DataSourceConfig{
|
||||
Name: name,
|
||||
Host: gconv.String(host),
|
||||
Port: int(gconv.Int(subMap["port"])),
|
||||
APIKey: gconv.String(subMap["apiKey"]),
|
||||
}
|
||||
|
||||
// 设置默认值
|
||||
if config.Port == 0 {
|
||||
config.Port = 7700
|
||||
}
|
||||
if config.Host == "" {
|
||||
config.Host = "http://localhost"
|
||||
}
|
||||
|
||||
// 可选：从配置读取超时时间
|
||||
if timeoutVal, hasTimeout := subMap["timeout"]; hasTimeout {
|
||||
config.Timeout = gconv.Duration(timeoutVal)
|
||||
} else {
|
||||
config.Timeout = 10 * time.Second
|
||||
}
|
||||
|
||||
// 注册数据源
|
||||
if err := m.RegisterDataSource(config); err != nil {
|
||||
glog.Errorf(ctx, "failed to register datasource [%s]: %v", name, err)
|
||||
if firstErr == nil {
|
||||
firstErr = err
|
||||
}
|
||||
continue
|
||||
}
|
||||
|
||||
// 连接数据源
|
||||
source, _ := m.GetDataSource(name)
|
||||
if err := source.Connect(ctx); err != nil {
|
||||
glog.Errorf(ctx, "failed to initialize datasource [%s]: %v", name, err)
|
||||
if firstErr == nil {
|
||||
firstErr = err
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
return firstErr
|
||||
}
|
||||
|
||||
// StartHealthCheck 启动健康检查
|
||||
func (m *DataSourceManager) StartHealthCheck() {
|
||||
if m.started {
|
||||
return
|
||||
}
|
||||
m.started = true
|
||||
go m.healthCheckLoop()
|
||||
}
|
||||
|
||||
// healthCheckLoop 健康检查循环
|
||||
func (m *DataSourceManager) healthCheckLoop() {
|
||||
ticker := time.NewTicker(30 * time.Second)
|
||||
defer ticker.Stop()
|
||||
|
||||
for {
|
||||
select {
|
||||
case <-m.ctx.Done():
|
||||
return
|
||||
case <-ticker.C:
|
||||
m.checkAndReconnect()
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
// checkAndReconnect 检查并重新连接
|
||||
func (m *DataSourceManager) checkAndReconnect() {
|
||||
m.mu.RLock()
|
||||
defer m.mu.RUnlock()
|
||||
|
||||
for name, source := range m.sources {
|
||||
if !source.IsConnected() {
|
||||
glog.Warningf(context.Background(), "datasource [%s] disconnected, attempting reconnect", name)
|
||||
|
||||
reconnectCtx, cancel := context.WithTimeout(context.Background(), 30*time.Second)
|
||||
defer cancel()
|
||||
|
||||
if err := source.Reconnect(reconnectCtx); err != nil {
|
||||
glog.Errorf(reconnectCtx, "datasource [%s] reconnect failed: %v", name, err)
|
||||
} else {
|
||||
glog.Infof(reconnectCtx, "✅ datasource [%s] reconnected successfully", name)
|
||||
}
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
// CloseAll 关闭所有数据源
|
||||
func (m *DataSourceManager) CloseAll(ctx context.Context) error {
|
||||
m.cancel()
|
||||
|
||||
m.mu.RLock()
|
||||
defer m.mu.RUnlock()
|
||||
|
||||
var lastErr error
|
||||
for name, source := range m.sources {
|
||||
if err := source.Close(ctx); err != nil {
|
||||
glog.Errorf(ctx, "failed to close datasource [%s]: %v", name, err)
|
||||
lastErr = err
|
||||
}
|
||||
}
|
||||
return lastErr
|
||||
}
|
||||
|
||||
// setupGracefulShutdown 设置优雅关闭
|
||||
func setupGracefulShutdown() {
|
||||
go func() {
|
||||
sigCh := make(chan os.Signal, 1)
|
||||
signal.Notify(sigCh, syscall.SIGINT, syscall.SIGTERM)
|
||||
<-sigCh
|
||||
|
||||
ctx, cancel := context.WithTimeout(context.Background(), 10*time.Second)
|
||||
defer cancel()
|
||||
|
||||
glog.Info(ctx, "🔄 Shutting down Meilisearch connections...")
|
||||
if err := manager.CloseAll(ctx); err != nil {
|
||||
glog.Errorf(ctx, "❌ Failed to close Meilisearch connections: %v", err)
|
||||
} else {
|
||||
glog.Info(ctx, "✅ Meilisearch connections closed successfully")
|
||||
}
|
||||
}()
|
||||
}
|
||||
721
full-text-search/meilisearch/meilisearch.go
Normal file
721
full-text-search/meilisearch/meilisearch.go
Normal file
@@ -0,0 +1,721 @@
|
||||
// =============================================================================
|
||||
// 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
|
||||
}
|
||||
109
full-text-search/meilisearch/types.go
Normal file
109
full-text-search/meilisearch/types.go
Normal file
@@ -0,0 +1,109 @@
|
||||
// =============================================================================
|
||||
// Meilisearch 数据结构定义
|
||||
// =============================================================================
|
||||
|
||||
package meilisearch
|
||||
|
||||
// SearchParams 搜索参数
|
||||
type SearchParams struct {
|
||||
Query string // 搜索查询字符串
|
||||
Page int64 // 页码（从1开始）
|
||||
Limit int64 // 每页数量
|
||||
Sort []string // 排序字段，如 ["createdAt:desc", "price:asc"]
|
||||
Filter string // 过滤条件
|
||||
SearchableAttributes string // 可搜索字段
|
||||
AttributesToRetrieve []string // 返回字段
|
||||
Facets []string // 聚合字段
|
||||
HitsPerPage int // 每页命中数
|
||||
}
|
||||
|
||||
// SearchResult 搜索结果
|
||||
type SearchResult struct {
|
||||
Hits []map[string]interface{} `json:"hits"`
|
||||
EstimatedTotalHits int64 `json:"estimatedTotalHits"`
|
||||
Limit int `json:"limit"`
|
||||
Offset int `json:"offset"`
|
||||
ProcessingTimeMs int `json:"processingTimeMs"`
|
||||
Query string `json:"query"`
|
||||
FacetDistribution map[string]interface{} `json:"facetDistribution,omitempty"`
|
||||
}
|
||||
|
||||
// IndexConfig 索引配置
|
||||
type IndexConfig struct {
|
||||
UID string // 索引唯一标识
|
||||
PrimaryKey string // 主键字段名
|
||||
}
|
||||
|
||||
// IndexSettings 索引设置
|
||||
type IndexSettings struct {
|
||||
SearchableAttributes []string `json:"searchableAttributes,omitempty"`
|
||||
DisplayedAttributes []string `json:"displayedAttributes,omitempty"`
|
||||
FilterableAttributes []string `json:"filterableAttributes,omitempty"`
|
||||
SortableAttributes []string `json:"sortableAttributes,omitempty"`
|
||||
RankingRules []string `json:"rankingRules,omitempty"`
|
||||
StopWords []string `json:"stopWords,omitempty"`
|
||||
Synonyms map[string][]string `json:"synonyms,omitempty"`
|
||||
DistinctAttribute string `json:"distinctAttribute,omitempty"`
|
||||
TypoTolerance *TypoTolerance `json:"typoTolerance,omitempty"`
|
||||
Pagination *Pagination `json:"pagination,omitempty"`
|
||||
Faceting *Faceting `json:"faceting,omitempty"`
|
||||
}
|
||||
|
||||
// TypoTolerance 拼写容错设置
|
||||
type TypoTolerance struct {
|
||||
Enabled bool `json:"enabled"`
|
||||
MinWordSizeForTypos map[string]int `json:"minWordSizeForTypos"`
|
||||
DisableOnWords []string `json:"disableOnWords"`
|
||||
DisableOnAttributes []string `json:"disableOnAttributes"`
|
||||
}
|
||||
|
||||
// Pagination 分页设置
|
||||
type Pagination struct {
|
||||
MaxTotalHits int `json:"maxTotalHits"`
|
||||
}
|
||||
|
||||
// Faceting 聚合设置
|
||||
type Faceting struct {
|
||||
MaxValuesPerFacet int `json:"maxValuesPerFacet"`
|
||||
}
|
||||
|
||||
// TaskResult 任务结果
|
||||
type TaskResult struct {
|
||||
TaskUID int64 `json:"taskUid"`
|
||||
Status string `json:"status"`
|
||||
Type string `json:"type"`
|
||||
Details map[string]interface{} `json:"details,omitempty"`
|
||||
Error string `json:"error,omitempty"`
|
||||
}
|
||||
|
||||
// Document 文档
|
||||
type Document map[string]interface{}
|
||||
|
||||
// 文档操作符
|
||||
const (
|
||||
OperatorEquals = "="
|
||||
OperatorNotEquals = "!="
|
||||
OperatorGreaterThan = ">"
|
||||
OperatorGreaterOrEqual = ">="
|
||||
OperatorLessThan = "<"
|
||||
OperatorLessOrEqual = "<="
|
||||
OperatorAnd = "AND"
|
||||
OperatorOr = "OR"
|
||||
OperatorNot = "NOT"
|
||||
)
|
||||
|
||||
// 排序顺序
|
||||
const (
|
||||
SortAsc = "asc"
|
||||
SortDesc = "desc"
|
||||
)
|
||||
|
||||
// 排名规则
|
||||
const (
|
||||
RankingRuleWords = "words"
|
||||
RankingRuleTypo = "typo"
|
||||
RankingRuleProximity = "proximity"
|
||||
RankingRuleAttribute = "attribute"
|
||||
RankingRuleSort = "sort"
|
||||
RankingRuleExactness = "exactness"
|
||||
)
|
||||
Reference in New Issue
Block a user