web_huangguo_server/web/service/srv_search/updater.go

package srv_search

import (
	"context"
	"fmt"
	"go.mongodb.org/mongo-driver/bson"
	"go.mongodb.org/mongo-driver/bson/primitive"
	"go.mongodb.org/mongo-driver/mongo"
	"go.mongodb.org/mongo-driver/mongo/options"
	"strconv"
	"videoweb/pub/common/log"
	"videoweb/pub/common/srvdef"
	"videoweb/pub/common/utils"
	"videoweb/pub/esmanager"
	"videoweb/pub/models/esmod"
	"videoweb/pub/mongos"
)

// resetAllDocsToES 重设所有文档到数据库
func resetAllDocsToES(ctx context.Context, param *Param) *srvdef.SrvErr {
	log.Info("srv_search:resetAllDocsToES...", log.Any("ESIndex", param.EsIndex))
	defer log.Info("srv_search:startActress!!!")
	//删除所有记录
	delAllFromES(param.EsIndex)

	var pageNum int64 = 1
	var pageSize int64 = 200
	updSize := 0
	for {
		findOpt := constructFindOption(pageNum, pageSize)
		docs, serr := findManyFromDb(ctx, param, findOpt)
		if serr != nil {
			return serr
		}
		if len(docs) == 0 {
			break
		}
		updSize = updSize + len(docs)
		updateESData, serr := sliceMapToJsonMap(docs, param)
		log.Info("srv_search:resetAllDocsToES", log.Any("updateESData", updateESData))
		if serr = batchAddToES(param.EsIndex, updateESData); serr != nil {
			return serr
		}
		pageNum++
		log.Info("resetAllToES>>>>>>>>>>>>>>>>>>", log.Any("updateSize:", updSize))
	}
	return nil
}

// constructMatch 构建match
func constructMatch(param *Param) bson.M {
	updateFilter := bson.A{}
	for _, fieldName := range param.CollFields {
		filter := bson.M{"updateDescription.updatedFields." + fieldName: bson.M{"$exists": true}}
		updateFilter = append(updateFilter, filter)
	}
	match := bson.M{"$or": bson.A{
		bson.M{"operationType": bson.M{"$in": []string{"delete", "insert", "replace"}}},
		bson.M{"$and": bson.A{
			bson.M{"operationType": bson.M{"$in": []string{"update"}}},
			bson.M{"$or": updateFilter}},
		}}}
	return match
}

// constructActtressOption 查找选项
func constructFindOption(pageNum int64, pageSize int64) *options.FindOptions {
	opt := &options.FindOptions{}
	opt.SetSort(bson.M{"createdAt": -1})
	skip := (pageNum - 1) * pageSize
	opt.SetSkip(skip)
	opt.SetLimit(pageSize)
	return opt
}

// AddMultiDataToES 添加一批Ex
func AddMultiDataToES(esIndex string, esvs esmanager.M) *srvdef.SrvErr {
	if err := esmanager.Instance.Bulk(esIndex, esvs); err != nil {
		info := fmt.Sprintf("(srv_search:AddMultiDataToES) ES Bulk failed! err:%+v:", err)
		log.Warn(info)
		return srvdef.New(srvdef.ErrElsOperError, "", "")
	}
	return nil
}

// findManyFromDb 查找一批数据
func findManyFromDb(ctx context.Context, param *Param,
	opts ...*options.FindOptions) ([]bson.M, *srvdef.SrvErr) {
	docs := make([]bson.M, 0)
	if err := param.Coll.Find(ctx, &docs, param.CollFilter, opts...); err != nil {
		log.Warn("srv_search:findManyFromDb db.Find failed", log.E(err), log.Any("filter", param.CollFilter))
		return nil, srvdef.New(srvdef.ErrDbQueryError, "", "")
	}
	return docs, nil
}

// sliceMapToJsonMap 从map切片构建Map对象
func sliceMapToJsonMap(docs []bson.M, param *Param) (esmanager.M, *srvdef.SrvErr) {
	esMap := make(esmanager.M)
	for _, doc := range docs {
		fieldMap := make(esmanager.M)
		for key, value := range doc {
			if utils.IncludeString(param.CollFields, key) {
				fieldMap[key] = value
			} else if key == "_id" {
				fieldMap["id"] = value
			}
		}
		if id, ok := doc["_id"].(int64); ok {
			strId := strconv.FormatInt(id, 10)
			esMap[strId] = fieldMap
		} else {
			log.Warn("srv_search:sliceMapToJsonMap Key doesn't exist", log.Any("keyId", "_id"))
		}
	}
	return esMap, nil
}

// getInfoFromChangeData 从改变流中获取数据信息
func getInfoFromChangeData(changeData bson.M, param *Param) (
	string, *esmod.ESTimestamp, string, esmanager.M, *srvdef.SrvErr) {
	operationType, ok := changeData["operationType"].(string)
	if ok == false {
		log.Error("srv_search:getInfoFromChangeData operationType type error")
		return "", nil, "", nil, srvdef.New(srvdef.ErrData, "", "")
	}
	ptimestamp, ok := changeData["clusterTime"].(primitive.Timestamp)
	if ok == false {
		log.Error("srv_search:getInfoFromChangeData clusterTime type error")
		return "", nil, "", nil, srvdef.New(srvdef.ErrData, "", "")
	}
	estimestamp := &esmod.ESTimestamp{T: ptimestamp.T, I: ptimestamp.I}
	documentKey, ok := changeData["documentKey"].(primitive.M)
	if ok == false {
		log.Error("srv_search:getInfoFromChangeData documentKey type error")
		return "", nil, "", nil, srvdef.New(srvdef.ErrData, "", "")
	}
	docKey, ok := documentKey["_id"].(int64)
	if ok == false {
		log.Error("srv_search:getInfoFromChangeData documentKey[_id] type error")
		return "", nil, "", nil, srvdef.New(srvdef.ErrData, "", "")
	}
	strId := strconv.FormatInt(docKey, 10)
	esDocMap := make(esmanager.M)
	if changeData["fullDocument"] != nil {
		docMap, ok := changeData["fullDocument"].(primitive.M)
		if ok == false {
			log.Error("srv_search:getInfoFromChangeData fullDocument type error")
			return "", nil, "", nil, srvdef.New(srvdef.ErrData, "", "")
		}

		for key, value := range docMap {
			if utils.IncludeString(param.CollFields, key) {
				esDocMap[key] = value
			} else if key == "_id" {
				esDocMap["id"] = value
			}
		}
	}
	return operationType, estimestamp, strId, esDocMap, nil
}

// delAllFromES 删除所有数据
func delAllFromES(esIndex string) *srvdef.SrvErr {
	query := esmanager.M{
		"query": esmanager.M{
			"match_all": esmanager.M{},
		},
	}
	if err := esmanager.Instance.DelMany([]string{esIndex}, query); err != nil {
		log.Warn(fmt.Sprintf("(srv_search:DelAllFromEs) Es DelMany failed! err:%+v:", err),
			log.Any("query", query))
		return srvdef.New(srvdef.ErrElsOperError, "", "")
	}
	return nil
}

// batchAddToES 添加一批数据到ES
func batchAddToES(esIndex string, esvs esmanager.M) *srvdef.SrvErr {
	log.Info("srv_search:resetAllDocsToES", log.Any("esIndex", esIndex), log.Any("esvs", esvs))
	if err := esmanager.Instance.Bulk(esIndex, esvs); err != nil {
		info := fmt.Sprintf("(srv_search:batchAddToES) ES Bulk failed! err:%+v:", err)
		log.Warn(info)
		return srvdef.New(srvdef.ErrElsOperError, "", "")
	}
	return nil
}

// batchDelFromES 删除一批数据从ES
func batchDelFromES(esIndex string, ids []string) *srvdef.SrvErr {
	query := esmanager.M{
		"query": esmanager.M{
			"bool": esmanager.M{
				"must": esmanager.A{
					{
						"ids": esmanager.M{
							"values": ids,
						},
					},
				},
			},
		},
	}
	if err := esmanager.Instance.DelMany([]string{esIndex}, query); err != nil {
		info := fmt.Sprintf(
			"(srv_search:batchDelFromES) Es DelMany failed! err:%+v:", err)
		log.Warn(info)
		return srvdef.New(srvdef.ErrElsOperError, "", "")
	}
	return nil
}

// watchOpLog 监控表变化
func watchOpLog(ctx context.Context, db *mongos.Collection, pipeline []bson.M,
	opt *options.ChangeStreamOptions) (*mongo.ChangeStream, *srvdef.SrvErr) {
	changeStream, err := db.WatchOpLog(ctx, pipeline, opt)
	if err != nil {
		log.Warn("srv_search:watchOpLog db.WatchOpLog failed",
			log.E(err), log.Any("pipeline", pipeline))
		return nil, srvdef.New(srvdef.ErrDbQueryError, "", "")
	}
	return changeStream, nil
}