package esmod import ( "context" "go.mongodb.org/mongo-driver/bson/primitive" "time" "videoweb/pub/common/log" "videoweb/pub/common/srvdef" "go.mongodb.org/mongo-driver/bson" "go.mongodb.org/mongo-driver/mongo" "go.mongodb.org/mongo-driver/mongo/options" ) // InsertOneESRecord 插入ES操作记录 func InsertOneESRecord(ctx context.Context, esRecord *ESRecord) *srvdef.SrvErr { if _, err := db.InsertOne(ctx, esRecord); err != nil { log.Warn("esmod:InsertOneESRecord db.InsertOne failed", log.E(err)) return srvdef.New(srvdef.ErrDbInsertError, "", "") } return nil } // DelOneESRecord 删除一个ES操作记录 func DelOneESRecord(ctx context.Context, filter bson.M) ( int64, *srvdef.SrvErr) { count, err := db.DeleteOne(ctx, filter) if err != nil { log.Warn("esmod:DelOnAppTag db.DeleteOne failed", log.E(err), log.Any("filter", filter)) return 0, srvdef.New(srvdef.ErrDbDeleteError, "", "") } return count, nil } // FindOneESRecord 查找一个ES操作记录 func FindOneESRecord(ctx context.Context, filter bson.M) ( *ESRecord, *srvdef.SrvErr) { var esRecord ESRecord if err := db.FindOneEx( ctx, &esRecord, filter); err != nil { if err != mongo.ErrNoDocuments { log.Info("esmod:FindOnESRecord db.FindOne failed", log.E(err), log.Any("filter", filter)) return nil, srvdef.New(srvdef.ErrDbQueryError, "", "") } return nil, srvdef.New(srvdef.ErrNoData, "", "") } return &esRecord, nil } // CountESRecords 统计ES操作记录 func CountESRecords(ctx context.Context, filter bson.M) ( int64, *srvdef.SrvErr) { count, err := db.Count(ctx, filter) if err != nil { log.Warn("esmod:CountMediaCats db.Count failed", log.E(err), log.Any("filter", filter)) return 0, srvdef.New(srvdef.ErrDbQueryError, "", "") } return count, nil } // UpdateOneESRecord 更新ES操作记录 func UpdateOneESRecord(ctx context.Context, filter bson.M, update bson.M) *srvdef.SrvErr { result, err := db.UpdateOne(ctx, filter, update) if err != nil { log.Warn("esmod:UpdateOnESRecord db.UpdateOne failed", log.E(err), log.Any("filter", filter)) return srvdef.New(srvdef.ErrDbUpdateError, "", "") } if result.MatchedCount != 1 { log.Warn("esmod:UpdateOnESRecord db.UpdateOne failed", log.E(err), log.Any("filter", filter)) return srvdef.New(srvdef.ErrMediaRecordNotExsit, "", "") } return nil } // UpsertOneESRecord 插入或更新一条ES操作记录 func UpsertOneESRecord(ctx context.Context, filter bson.M, update bson.M) *srvdef.SrvErr { if _, err := db.UpsertOne(ctx, filter, update); err != nil { log.Info("esmod:UpsertOneESRecord db.InsertOne failed", log.E(err), log.Any("filter", filter)) return srvdef.New(srvdef.ErrDbUpdateError, "", "") } return nil } // UpdateManyESRecord 更新一批ES操作记录 func UpdateManyESRecord(ctx context.Context, filter bson.M, update bson.M) *srvdef.SrvErr { result, err := db.UpdateMMany(ctx, filter, update) if err != nil { log.Warn("esmod:UpdateManyMediaCat db.UpdateMMany failed", log.E(err), log.Any("filter", filter)) return srvdef.New(srvdef.ErrDbUpdateError, "", "") } if result.MatchedCount <= 0 { log.Warn("esmod:UpdateManyMediaCat db.UpdateMMany failed", log.E(err), log.Any("filter", filter)) //return srvdef.New(srvdef.ErrMediaTagNotExsit, "", "") } return nil } // FindManyESRecord 查找一批ES操作记录 func FindManyESRecord(ctx context.Context, filter bson.M, opts ...*options.FindOptions) (ESRecordList, *srvdef.SrvErr) { data := make(ESRecordList, 0) if err := db.Find(ctx, &data, filter, opts...); err != nil { log.Warn("esmod:FindManyESRecord db.Find failed", log.E(err), log.Any("filter", filter)) return nil, srvdef.New(srvdef.ErrDbQueryError, "", "") } return data, nil } func GetRecordTimeByESIndex(ctx context.Context, esIndex string) (*ESRecord, *srvdef.SrvErr) { filter := bson.M{FN.ESRecord.EsIndex: esIndex} esRecord, serr := FindOneESRecord(ctx, filter) if serr != nil { if serr.Code != srvdef.ErrNoData { return nil, serr } return nil, nil } return esRecord, nil } func SetOneEsRecordTime(ctx context.Context, esIndex string, recordTime time.Time) *srvdef.SrvErr { filter := bson.M{FN.ESRecord.EsIndex: esIndex} upsertDoc := bson.M{ "$set": bson.M{ FN.ESRecord.RecordTime: recordTime, FN.ESRecord.UpdatedAt: time.Now(), }, "$setOnInsert": bson.M{ FN.ESRecord.Id: primitive.NewObjectID(), FN.ESRecord.CreatedAt: time.Now(), }, } serr := UpsertOneESRecord(ctx, filter, upsertDoc) return serr }