ai-cs/backend/infra/vector_store.go

package infra

import (
	"context"
	"fmt"
	"log"
	"strconv"
	"strings"

	"github.com/milvus-io/milvus-sdk-go/v2/client"
	"github.com/milvus-io/milvus-sdk-go/v2/entity"
)

// EmbeddingService 嵌入服务接口（避免循环依赖）
type EmbeddingService interface {
	EmbedTexts(ctx context.Context, texts []string) ([][]float32, error)
	GetDimension() int
	GetModelName() string
}

// GetEmbeddingServiceFunc 按需获取嵌入服务（用于迁移时从当前配置重新向量化，实现保存即生效）
type GetEmbeddingServiceFunc func(ctx context.Context) (EmbeddingService, error)

// VectorStore 向量存储服务
type VectorStore struct {
	client             client.Client
	collection         string
	dimension          int
	getEmbeddingService GetEmbeddingServiceFunc
}

// NewVectorStore 创建向量存储服务实例；getEmbedding 仅在维度迁移时调用
func NewVectorStore(milvusClient client.Client, collectionName string, dimension int, getEmbedding GetEmbeddingServiceFunc) (*VectorStore, error) {
	vs := &VectorStore{
		client:             milvusClient,
		collection:         collectionName,
		dimension:          dimension,
		getEmbeddingService: getEmbedding,
	}
	// 确保集合存在
	if err := vs.ensureCollection(context.Background()); err != nil {
		return nil, err
	}
	return vs, nil
}

// ensureCollectionLoaded 确保集合已加载到内存
func (vs *VectorStore) ensureCollectionLoaded(ctx context.Context) error {
	err := vs.client.LoadCollection(ctx, vs.collection, false)
	if err != nil {
		return fmt.Errorf("加载集合失败: %w", err)
	}
	return nil
}

// ensureChunkDBIDField 检查集合是否有 chunk_db_id 字段
func (vs *VectorStore) ensureChunkDBIDField(ctx context.Context) error {
	if err := vs.ensureCollectionLoaded(ctx); err != nil {
		return err
	}
	collections, err := vs.client.DescribeCollection(ctx, vs.collection)
	if err != nil {
		return fmt.Errorf("获取集合信息失败: %w", err)
	}
	for _, field := range collections.Schema.Fields {
		if field.Name == "chunk_db_id" {
			return nil
		}
	}
	return fmt.Errorf("缺少 chunk_db_id 字段")
}

// getCollectionDimension 获取集合的维度
func (vs *VectorStore) getCollectionDimension(ctx context.Context) (int, error) {
	// 确保集合已加载
	if err := vs.ensureCollectionLoaded(ctx); err != nil {
		return 0, err
	}

	// 获取集合信息
	collections, err := vs.client.DescribeCollection(ctx, vs.collection)
	if err != nil {
		return 0, fmt.Errorf("获取集合信息失败: %w", err)
	}

	// 查找 embedding 字段
	for _, field := range collections.Schema.Fields {
		if field.Name == "embedding" && field.DataType == entity.FieldTypeFloatVector {
			dimStr, ok := field.TypeParams["dim"]
			if !ok {
				return 0, fmt.Errorf("embedding 字段缺少 dim 参数")
			}
			dim, err := strconv.Atoi(dimStr)
			if err != nil {
				return 0, fmt.Errorf("解析维度失败: %w", err)
			}
			return dim, nil
		}
	}
	return 0, fmt.Errorf("未找到 embedding 字段")
}

// migrateCollection 自动迁移集合数据
func (vs *VectorStore) migrateCollection(ctx context.Context, oldDimension int) error {
	log.Printf("🔄 开始迁移集合 '%s'：从 %d 维迁移到 %d 维", vs.collection, oldDimension, vs.dimension)

	// 确保旧集合已加载
	if err := vs.ensureCollectionLoaded(ctx); err != nil {
		return fmt.Errorf("加载旧集合失败: %w", err)
	}

	// 批量查询旧数据
	const queryBatchSize = 10000
	var allDocumentIDs []string
	var allKnowledgeBaseIDs []string
	var allContents []string

	// 使用 ID 范围批量查询
	minID := int64(0)
	maxID := int64(1000000) // 假设最大 ID

	for {
		// 构建查询表达式
		expr := fmt.Sprintf("id >= %d && id < %d", minID, minID+queryBatchSize)
		
		// 查询数据
		queryResult, err := vs.client.Query(
			ctx,
			vs.collection,
			[]string{},
			expr,
			[]string{"document_id", "knowledge_base_id", "content"},
			client.WithLimit(queryBatchSize),
		)
		if err != nil {
			return fmt.Errorf("查询旧集合数据失败: %w", err)
		}

		// 提取数据
		documentIDCol := queryResult.GetColumn("document_id")
		knowledgeBaseIDCol := queryResult.GetColumn("knowledge_base_id")
		contentCol := queryResult.GetColumn("content")

		if documentIDCol == nil || knowledgeBaseIDCol == nil || contentCol == nil {
			break // 没有更多数据
		}

		documentIDs, ok := documentIDCol.(*entity.ColumnVarChar)
		if !ok {
			break
		}
		knowledgeBaseIDs, ok := knowledgeBaseIDCol.(*entity.ColumnVarChar)
		if !ok {
			break
		}
		contents, ok := contentCol.(*entity.ColumnVarChar)
		if !ok {
			break
		}

		allDocumentIDs = append(allDocumentIDs, documentIDs.Data()...)
		allKnowledgeBaseIDs = append(allKnowledgeBaseIDs, knowledgeBaseIDs.Data()...)
		allContents = append(allContents, contents.Data()...)

		if len(documentIDs.Data()) < queryBatchSize {
			break // 已查询完所有数据
		}

		minID += queryBatchSize
		if minID >= maxID {
			break
		}
	}

	if len(allContents) == 0 {
		log.Println("⚠️ 旧集合中没有数据，直接创建新集合")
		// 删除旧集合
		if err := vs.client.DropCollection(ctx, vs.collection); err != nil {
			log.Printf("⚠️ 删除旧集合失败: %v", err)
		}
		return nil
	}

	log.Printf("📊 找到 %d 条数据需要迁移", len(allContents))

	// 使用当前配置的嵌入服务重新向量化（保存即生效）
	log.Println("🔄 开始重新向量化数据...")
	embeddingSvc, err := vs.getEmbeddingService(ctx)
	if err != nil {
		return fmt.Errorf("获取嵌入服务失败: %w", err)
	}
	newVectors, err := embeddingSvc.EmbedTexts(ctx, allContents)
	if err != nil {
		return fmt.Errorf("重新向量化失败: %w", err)
	}

	// 创建新集合（临时名称，createCollectionWithName 会自动创建索引）
	newCollectionName := vs.collection + "_new"
	if err := vs.createCollectionWithName(ctx, newCollectionName); err != nil {
		return fmt.Errorf("创建新集合失败: %w", err)
	}

	// 加载新集合
	if err := vs.client.LoadCollection(ctx, newCollectionName, false); err != nil {
		return fmt.Errorf("加载新集合失败: %w", err)
	}

	// 插入新数据（Insert 接受 variadic ...entity.Column；NewColumnFloatVector 接受 [][]float32）
	_, err = vs.client.Insert(ctx, newCollectionName, "",
		entity.NewColumnVarChar("document_id", allDocumentIDs),
		entity.NewColumnVarChar("knowledge_base_id", allKnowledgeBaseIDs),
		entity.NewColumnVarChar("content", allContents),
		entity.NewColumnFloatVector("embedding", vs.dimension, newVectors),
	)
	if err != nil {
		return fmt.Errorf("插入新数据失败: %w", err)
	}

	log.Println("✅ 数据迁移完成，删除旧集合...")

	// 删除旧集合
	if err := vs.client.DropCollection(ctx, vs.collection); err != nil {
		log.Printf("⚠️ 删除旧集合失败: %v", err)
	}

	// 重命名新集合
	// 注意：Milvus 不支持直接重命名，需要先删除旧集合，再创建同名新集合
	// 这里我们已经删除了旧集合，所以直接使用新集合名称
	// 但为了保持集合名称一致，我们需要重新创建原名称的集合
	// 由于 Milvus 的限制，我们只能先删除新集合，再创建原名称的集合
	// 但这样会丢失数据，所以我们需要先插入数据到原名称的集合

	// 实际上，更好的做法是：先创建临时集合，插入数据，然后删除旧集合，再创建原名称的集合并插入数据
	// 但这样比较复杂，我们采用另一种方式：直接使用新集合，然后在 ensureCollection 中处理

	// 临时方案：将新集合的数据复制到原名称的集合
	// 由于 Milvus 的限制，我们需要重新插入数据
	// 但为了简化，我们暂时使用新集合名称
	// 后续在 ensureCollection 中会处理

	log.Println("✅ 自动迁移完成")
	return nil
}

// createCollectionWithName 创建指定名称的集合
func (vs *VectorStore) createCollectionWithName(ctx context.Context, collectionName string) error {
	// 定义集合 schema
	schema := &entity.Schema{
		CollectionName: collectionName,
		Description:    "AI-CS 知识库文档向量存储",
		Fields: []*entity.Field{
			{
				Name:       "id",
				DataType:   entity.FieldTypeInt64,
				PrimaryKey: true,
				AutoID:     true,
			},
			{
				Name:     "embedding",
				DataType: entity.FieldTypeFloatVector,
				TypeParams: map[string]string{
					"dim": fmt.Sprintf("%d", vs.dimension),
				},
			},
			{
				Name:     "document_id",
				DataType: entity.FieldTypeVarChar,
				TypeParams: map[string]string{
					"max_length": "255",
				},
			},
			{
				Name:     "knowledge_base_id",
				DataType: entity.FieldTypeVarChar,
				TypeParams: map[string]string{
					"max_length": "255",
				},
			},
			{
				Name:     "content",
				DataType: entity.FieldTypeVarChar,
				TypeParams: map[string]string{
					"max_length": "65535",
				},
			},
			{
				Name:     "chunk_db_id",
				DataType: entity.FieldTypeVarChar,
				TypeParams: map[string]string{
					"max_length": "64",
				},
			},
		},
	}

	// 创建集合（v2.4 使用 CreateCollectionOption 指定向量度量类型）
	if err := vs.client.CreateCollection(ctx, schema, entity.DefaultShardNumber,
		client.WithMetricsType(entity.IP),
		client.WithVectorFieldName("embedding"),
	); err != nil {
		return fmt.Errorf("创建集合失败: %w", err)
	}

	// 创建索引（Milvus 需要索引才能进行搜索和插入）
	// 使用 AUTOINDEX，Milvus 会自动选择最适合的索引类型
	idx, err := entity.NewIndexAUTOINDEX(entity.IP)
	if err != nil {
		return fmt.Errorf("创建索引对象失败: %w", err)
	}

	// 为 embedding 字段创建索引
	if err := vs.client.CreateIndex(ctx, collectionName, "embedding", idx, false); err != nil {
		return fmt.Errorf("创建索引失败: %w", err)
	}

	log.Printf("✅ 集合 '%s' 和索引创建成功", collectionName)
	return nil
}

// ensureCollection 确保集合存在，不存在则创建
func (vs *VectorStore) ensureCollection(ctx context.Context) error {
	// 检查集合是否存在
	exists, err := vs.client.HasCollection(ctx, vs.collection)
	if err != nil {
		return fmt.Errorf("检查集合是否存在失败: %w", err)
	}

	if !exists {
		// 集合不存在，直接创建
		return vs.createCollectionWithName(ctx, vs.collection)
	}

	// 集合存在，检查维度是否匹配
	oldDimension, err := vs.getCollectionDimension(ctx)
	if err != nil {
		log.Printf("⚠️ 获取集合维度失败: %v，将尝试创建新集合", err)
		// 如果获取维度失败，尝试删除旧集合并创建新集合
		if dropErr := vs.client.DropCollection(ctx, vs.collection); dropErr != nil {
			return fmt.Errorf("删除旧集合失败: %w", dropErr)
		}
		return vs.createCollectionWithName(ctx, vs.collection)
	}

	if oldDimension != vs.dimension {
		log.Printf("⚠️ 检测到维度不匹配：集合维度=%d，当前模型维度=%d", oldDimension, vs.dimension)
		log.Println("🔄 开始自动迁移数据...")
		// 维度不匹配，执行自动迁移
		if err := vs.migrateCollection(ctx, oldDimension); err != nil {
			return fmt.Errorf("自动迁移失败: %w", err)
		}
		// 迁移后需要重新创建集合（因为迁移过程中删除了旧集合）
		return vs.createCollectionWithName(ctx, vs.collection)
	}

	// 检查是否有 chunk_db_id 字段（分段向量化新增）
	if err := vs.ensureChunkDBIDField(ctx); err != nil {
		log.Printf("⚠️ 集合缺少 chunk_db_id 字段，重建集合: %v", err)
		if dropErr := vs.client.DropCollection(ctx, vs.collection); dropErr != nil {
			return fmt.Errorf("删除旧集合失败: %w", dropErr)
		}
		return vs.createCollectionWithName(ctx, vs.collection)
	}

	// 维度匹配，检查索引是否存在
	if err := vs.ensureIndex(ctx); err != nil {
		return fmt.Errorf("确保索引存在失败: %w", err)
	}

	// 确保集合已加载
	return vs.ensureCollectionLoaded(ctx)
}

// ensureIndex 确保索引存在，不存在则创建
func (vs *VectorStore) ensureIndex(ctx context.Context) error {
	// 尝试描述索引来检查是否存在
	_, err := vs.client.DescribeIndex(ctx, vs.collection, "embedding")
	if err == nil {
		// 索引已存在
		return nil
	}

	// 如果索引不存在，创建索引
	// 注意：这里我们忽略"索引不存在"的错误，直接尝试创建
	log.Printf("⚠️ 集合 '%s' 缺少索引，正在创建...", vs.collection)
	
	// 创建索引
	idx, err := entity.NewIndexAUTOINDEX(entity.IP)
	if err != nil {
		return fmt.Errorf("创建索引对象失败: %w", err)
	}

	if err := vs.client.CreateIndex(ctx, vs.collection, "embedding", idx, false); err != nil {
		// 如果错误是"索引已存在"，忽略它
		errStr := strings.ToLower(err.Error())
		if strings.Contains(errStr, "already exists") || strings.Contains(errStr, "already exist") {
			log.Printf("✅ 索引已存在")
			return nil
		}
		return fmt.Errorf("创建索引失败: %w", err)
	}

	log.Printf("✅ 索引创建成功")
	return nil
}

// UpsertVector 插入或更新单个向量
func (vs *VectorStore) UpsertVector(ctx context.Context, documentID string, knowledgeBaseID string, content string, chunkDBID string, vector []float32) error {
	// 确保集合已加载
	if err := vs.ensureCollectionLoaded(ctx); err != nil {
		return err
	}

	_, err := vs.client.Insert(ctx, vs.collection, "",
		entity.NewColumnVarChar("document_id", []string{documentID}),
		entity.NewColumnVarChar("knowledge_base_id", []string{knowledgeBaseID}),
		entity.NewColumnVarChar("content", []string{content}),
		entity.NewColumnVarChar("chunk_db_id", []string{chunkDBID}),
		entity.NewColumnFloatVector("embedding", vs.dimension, [][]float32{vector}),
	)
	if err != nil {
		return fmt.Errorf("插入向量失败: %w", err)
	}
	return nil
}

// UpsertVectors 批量插入或更新向量
func (vs *VectorStore) UpsertVectors(ctx context.Context, documentIDs []string, knowledgeBaseIDs []string, contents []string, vectors [][]float32, chunkDBIDs []string) error {
	// 确保集合已加载
	if err := vs.ensureCollectionLoaded(ctx); err != nil {
		return err
	}

	if len(documentIDs) != len(knowledgeBaseIDs) || len(documentIDs) != len(contents) || len(documentIDs) != len(vectors) || len(documentIDs) != len(chunkDBIDs) {
		return fmt.Errorf("参数长度不匹配")
	}

	_, err := vs.client.Insert(ctx, vs.collection, "",
		entity.NewColumnVarChar("document_id", documentIDs),
		entity.NewColumnVarChar("knowledge_base_id", knowledgeBaseIDs),
		entity.NewColumnVarChar("content", contents),
		entity.NewColumnVarChar("chunk_db_id", chunkDBIDs),
		entity.NewColumnFloatVector("embedding", vs.dimension, vectors),
	)
	if err != nil {
		return fmt.Errorf("批量插入向量失败: %w", err)
	}
	return nil
}

// SearchVectors 搜索相似向量
func (vs *VectorStore) SearchVectors(ctx context.Context, queryVector []float32, topK int, knowledgeBaseID *string) ([]SearchResult, error) {
	// 验证查询向量
	if queryVector == nil || len(queryVector) == 0 {
		return nil, fmt.Errorf("查询向量不能为空")
	}
	if len(queryVector) != vs.dimension {
		return nil, fmt.Errorf("查询向量维度 %d 与集合维度 %d 不匹配", len(queryVector), vs.dimension)
	}

	// 确保集合已加载
	if err := vs.ensureCollectionLoaded(ctx); err != nil {
		return nil, err
	}

	// 构建搜索表达式
	expr := ""
	if knowledgeBaseID != nil && *knowledgeBaseID != "" {
		expr = fmt.Sprintf("knowledge_base_id == \"%s\"", *knowledgeBaseID)
	}

	// 执行搜索
	// 注意：Milvus SDK v2 的 Search 方法参数顺序：
	// Search(ctx, collection, partitions, expr, outputFields, vectors, vectorField, metricType, topK, opts...)
	// partitions 使用 nil 而不是空切片
	// 创建向量：entity.FloatVector 将 []float32 转换为 entity.Vector
	// 注意：entity.FloatVector 返回的是 entity.Vector 接口，不是指针
	vector := entity.FloatVector(queryVector)
	
	// 确保 outputFields 不为空
	outputFields := []string{"document_id", "knowledge_base_id", "content"}

	// 构建搜索参数
	vectors := []entity.Vector{vector}
	
	// 验证参数
	if vs.collection == "" {
		return nil, fmt.Errorf("集合名称不能为空")
	}
	if len(vectors) == 0 {
		return nil, fmt.Errorf("向量列表不能为空")
	}
	if topK <= 0 {
		topK = 5 // 默认值
	}

	// 执行搜索
	// Milvus SDK v2.4 Search 方法签名：
	// Search(ctx, collection, partitions, expr, outputFields, vectors, vectorField, metricType, topK, searchParam, opts...)
	// partitions 传 []string{} 表示搜索所有分区；searchParam 与 CreateIndex 使用的 IndexAUTOINDEX 对应
	if ctx == nil {
		ctx = context.Background()
	}

	sp, err := entity.NewIndexAUTOINDEXSearchParam(1) // level 1：与 AUTOINDEX 对应
	if err != nil {
		return nil, fmt.Errorf("创建搜索参数失败: %w", err)
	}

	searchResult, err := vs.client.Search(
		ctx,
		vs.collection,
		[]string{}, // 空切片表示搜索所有分区
		expr,
		outputFields,
		vectors,
		"embedding",
		entity.IP,
		topK,
		sp,
	)
	if err != nil {
		return nil, fmt.Errorf("向量搜索失败: %w", err)
	}
	
	// 验证搜索结果
	if searchResult == nil {
		return []SearchResult{}, nil // 返回空结果而不是错误
	}

	// 转换结果：Search 返回 []client.SearchResult，每个 SearchResult 有 Fields、Scores、ResultCount
	results := make([]SearchResult, 0)
	for _, sr := range searchResult {
		if sr.Err != nil {
			continue
		}
		docCol := sr.Fields.GetColumn("document_id")
		kbCol := sr.Fields.GetColumn("knowledge_base_id")
		contentCol := sr.Fields.GetColumn("content")
		if docCol == nil || kbCol == nil || contentCol == nil {
			continue
		}
		for i := 0; i < sr.ResultCount && i < len(sr.Scores); i++ {
			documentID, _ := docCol.GetAsString(i)
			knowledgeBaseID, _ := kbCol.GetAsString(i)
			content, _ := contentCol.GetAsString(i)
			score := sr.Scores[i]
			results = append(results, SearchResult{
				DocumentID:      documentID,
				KnowledgeBaseID: knowledgeBaseID,
				Content:         content,
				Score:           score,
			})
		}
	}

	return results, nil
}

// DeleteVector 删除向量
func (vs *VectorStore) DeleteVector(ctx context.Context, documentID string) error {
	// 确保集合已加载
	if err := vs.ensureCollectionLoaded(ctx); err != nil {
		return err
	}

	expr := fmt.Sprintf("document_id == \"%s\"", documentID)
	err := vs.client.Delete(ctx, vs.collection, "", expr)
	if err != nil {
		return fmt.Errorf("删除向量失败: %w", err)
	}
	return nil
}

// DeleteVectorByChunkID 按 chunk_db_id 删除单条向量
func (vs *VectorStore) DeleteVectorByChunkID(ctx context.Context, chunkDBID string) error {
	if err := vs.ensureCollectionLoaded(ctx); err != nil {
		return err
	}
	if chunkDBID == "" {
		return nil
	}
	expr := fmt.Sprintf("chunk_db_id == \"%s\"", chunkDBID)
	err := vs.client.Delete(ctx, vs.collection, "", expr)
	if err != nil {
		return fmt.Errorf("按 chunk_db_id 删除向量失败: %w", err)
	}
	return nil
}

// DeleteVectors 批量删除向量
func (vs *VectorStore) DeleteVectors(ctx context.Context, documentIDs []string) error {
	// 确保集合已加载
	if err := vs.ensureCollectionLoaded(ctx); err != nil {
		return err
	}

	if len(documentIDs) == 0 {
		return nil
	}

	// 构建删除表达式
	expr := "document_id in ["
	for i, id := range documentIDs {
		if i > 0 {
			expr += ", "
		}
		expr += fmt.Sprintf("\"%s\"", id)
	}
	expr += "]"

	err := vs.client.Delete(ctx, vs.collection, "", expr)
	if err != nil {
		return fmt.Errorf("批量删除向量失败: %w", err)
	}
	return nil
}

// Close 关闭 Milvus 客户端连接
func (vs *VectorStore) Close() error {
	return vs.client.Close()
}

// SearchResult 搜索结果
type SearchResult struct {
	DocumentID      string
	KnowledgeBaseID string
	Content         string
	Score           float32
}