package usermod import ( "context" "fmt" "time" "videoweb/pub/common/log" "videoweb/pub/common/srvdef" "videoweb/pub/mongos" "go.mongodb.org/mongo-driver/bson" "go.mongodb.org/mongo-driver/mongo" "go.mongodb.org/mongo-driver/mongo/options" ) var db *mongos.Collection func Init(ctx context.Context) { db = mongos.Shop.Coll(TN.User) indexes := []mongo.IndexModel{ { Keys: bson.D{ {Key: FN.User.Status, Value: 1}, }, }, { Keys: bson.D{ {Key: FN.User.AgentInviteCode, Value: -1}, }, }, { Keys: bson.D{ {Key: FN.User.InviteCode, Value: 1}, }, }, { Keys: bson.D{ {Key: FN.User.CreatedAt, Value: 1}, }, }, { Keys: bson.D{ {Key: FN.User.DevID, Value: 1}, }, }, { Keys: bson.D{ {Key: FN.User.Mac, Value: 1}, }, }, { Keys: bson.D{ {Key: FN.User.DevV2, Value: 1}, }, }, { Keys: bson.D{ {Key: FN.User.Recharge, Value: 1}, }, }, { Keys: bson.D{ {Key: FN.User.DistrictCode, Value: 1}, }, }, { Keys: bson.D{ {Key: FN.User.Type, Value: 1}, }, }, { Keys: bson.D{ {Key: FN.User.NickName, Value: "text"}, }, }, { Keys: bson.D{ {Key: FN.User.Status, Value: 1}, {Key: FN.User.Type, Value: 1}, }, }, // statcenter USER_BINDING:isRegister + updatedAt 范围 + sort,避免全表扫描 { Keys: bson.D{ {Key: FN.User.IsRegister, Value: 1}, {Key: FN.User.UpdatedAt, Value: 1}, }, Options: options.Index(). SetName("isRegister_1_updatedAt_1_bindingSync"). SetPartialFilterExpression(bson.M{ FN.User.IsRegister: true, FN.User.Email: bson.M{"$exists": true, "$gt": ""}, }), }, } if _, err := db.CreateIndex(ctx, indexes); err != nil { panic(fmt.Sprintf("(User:initUserIndex) Create User Index failed! err:%+v:", err)) } } // InsertOneUser 插入一个用户 func InsertOneUser(ctx context.Context, user *User) *srvdef.SrvErr { if _, err := db.InsertOne(ctx, user); err != nil { log.Warn("User:InsertOneUser db.InsertOne failed", log.E(err)) return srvdef.New(srvdef.ErrDbInsertError, "", "") } return nil } // FindOneUser 查找一个用户;opts 可用于调用方按需设置 projection,避免读取完整用户文档。 func FindOneUser(ctx context.Context, filter bson.M, opts ...*options.FindOneOptions) (*User, *srvdef.SrvErr) { var userInfo User if err := db.FindOneEx(ctx, &userInfo, filter, opts...); err != nil { if err != mongo.ErrNoDocuments { log.Warn("User:FindOneUser db.FindOne failed", log.E(err), log.Any("filter", filter)) return nil, srvdef.New(srvdef.ErrDbQueryError, "", "") } return nil, srvdef.New(srvdef.ErrUserNotExsit, "", "") } return &userInfo, nil } // DelOneUser 删除一个用户 func DelOneUser(ctx context.Context, filter bson.M) (int64, *srvdef.SrvErr) { count, err := db.DeleteOne(ctx, filter) if err != nil { log.Warn("User:DelOneUser db.DeleteOne failed", log.E(err), log.Any("filter", filter)) return 0, srvdef.New(srvdef.ErrDbDeleteError, "", "") } return count, nil } // UpdateOneUser 更新一个用户 func UpdateOneUser(ctx context.Context, filter bson.M, update bson.M) *srvdef.SrvErr { result, err := db.UpdateOne(ctx, filter, update) if err != nil { log.Warn("user:UpdateOneUser db.UpdateOne failed", log.E(err), log.Any("filter", filter)) return srvdef.New(srvdef.ErrDbUpdateError, "", "") } if result.MatchedCount != 1 { return srvdef.New(srvdef.ErrUserNotExsit, "", "") } return nil } // CountUsers 统计用户数量;无筛选条件时用 EstimatedCount,避免 user 全表 count 慢查 func CountUsers(ctx context.Context, filter bson.M) (int64, *srvdef.SrvErr) { if len(filter) == 0 { count, err := db.EstimatedCount(ctx) if err != nil { log.Warn("User:CountUsers db.EstimatedCount failed", log.E(err)) return 0, srvdef.New(srvdef.ErrDbQueryError, "", "") } return count, nil } count, err := db.Count(ctx, filter) if err != nil { log.Warn("User:CountUsers db.Count failed", log.E(err), log.Any("filter", filter)) return 0, srvdef.New(srvdef.ErrDbQueryError, "", "") } return count, nil } // UpdateManyUser 更新满足条件的用户 func UpdateManyUser(ctx context.Context, filter bson.M, update bson.M) *srvdef.SrvErr { _, err := db.UpdateMany(ctx, filter, update) if err != nil { log.Warn("user:UpdateManyUser db.UpdateMany failed", log.E(err), log.Any("filter", filter)) return srvdef.New(srvdef.ErrDbUpdateError, "", "") } return nil } // FindManyUser 查找一批用户数据 func FindManyUser(ctx context.Context, filter bson.M, opts ...*options.FindOptions) (UserList, *srvdef.SrvErr) { data := make(UserList, 0) if err := db.Find(ctx, &data, filter, opts...); err != nil { log.Warn("User:FindManyUser db.Find failed", log.E(err), log.Any("filter", filter)) return nil, srvdef.New(srvdef.ErrDbQueryError, "", "") } return data, nil } // FindManyUserWithBind 用自定义结构查找一批用户数据 func FindManyUserWithBind(ctx context.Context, bind interface{}, filter bson.M, opts ...*options.FindOptions) *srvdef.SrvErr { if err := db.Find(ctx, bind, filter, opts...); err != nil { log.Warn("User:FindManyUserWithBind db.Find failed", log.E(err), log.Any("filter", filter)) return srvdef.New(srvdef.ErrDbQueryError, "", "") } return nil } // AggregateUsers 聚合查找一批用户信息 func AggregateUsers(ctx context.Context, bind interface{}, pipeline []bson.M, opts ...*options.AggregateOptions) *srvdef.SrvErr { if err := db.Aggregate(ctx, bind, pipeline, opts...); err != nil { log.Warn("User:AggregateUsers db.Aggregate failed", log.E(err), log.Any("pipeline", pipeline)) return srvdef.New(srvdef.ErrDbQueryError, "", "") } return nil } // ObjectAggregate 聚合查找 func ObjectAggregate(ctx context.Context, bind interface{}, pipeline []bson.M) *srvdef.SrvErr { if err := db.ObjectAggregate(ctx, bind, pipeline); err != nil { log.Warn("User:ObjectAggregate db.ObjectAggregate failed", log.E(err), log.Any("pipeline", pipeline)) return srvdef.New(srvdef.ErrDbQueryError, "", "") } return nil } // UpdateRecharge 更新用户充值金额(正数为充值,负数为消费) func UpdateRecharge(ctx context.Context, userId, recharge, rechargeTotal int64) (*User, *srvdef.SrvErr) { userInfo := &User{} incDoc := bson.M{ FN.User.Recharge: recharge, FN.User.RechargeTotal: rechargeTotal, } if recharge < 0 { incDoc[FN.User.ConsumptionTotal] = -recharge } err := db.FindOneAndUpdate(ctx, &userInfo, bson.M{ FN.User.Id: userId, FN.User.Recharge: bson.M{"$gte": -recharge}, }, bson.M{ "$inc": incDoc, "$set": bson.M{FN.User.UpdatedAt: time.Now()}, }) if err != nil { log.Error("UpdateRecharge:FindOneAndUpdate failed", log.E(err)) return nil, srvdef.New(srvdef.DBUpdate, "数据更新出错", "") } if userInfo.Id <= 0 { return nil, srvdef.New(srvdef.ErrUserNotExsit, "", "") } return userInfo, nil } // UpdateScore 更新用户积分(正数加分,负数扣分) func UpdateScore(ctx context.Context, userId, score int64) (*User, *srvdef.SrvErr) { userInfo := &User{} incDoc := bson.M{FN.User.Score: score} filter := bson.M{FN.User.Id: userId} if score > 0 { incDoc[FN.User.ScoreTotal] = score } else { filter[FN.User.Score] = bson.M{"$gte": -score} } err := db.FindOneAndUpdate(ctx, &userInfo, filter, bson.M{ "$inc": incDoc, "$set": bson.M{FN.User.UpdatedAt: time.Now()}, }) if err != nil { log.Warn("usermod:UpdateScore FindOneAndUpdate failed", log.E(err), log.Any("filter", filter)) return nil, srvdef.New(srvdef.DBUpdate, "数据更新出错", "") } if userInfo.Id <= 0 { return nil, srvdef.New(srvdef.DBUpdate, "为0", "") } return userInfo, nil } type OnAddScore func(userId int64, add int64) // UpdateScoreWithEvent 更新用户积分并触发回调 func UpdateScoreWithEvent(ctx context.Context, userId, score int64, onAdd OnAddScore) (*User, *srvdef.SrvErr) { userInfo := &User{} incDoc := bson.M{FN.User.Score: score} filter := bson.M{FN.User.Id: userId} if score > 0 { incDoc[FN.User.ScoreTotal] = score } err := db.FindOneAndUpdate(ctx, &userInfo, filter, bson.M{ "$inc": incDoc, "$set": bson.M{FN.User.UpdatedAt: time.Now()}, }) if err != nil { log.Warn("usermod:UpdateScoreWithEvent FindOneAndUpdate failed", log.E(err), log.Any("filter", filter)) return nil, srvdef.New(srvdef.DBUpdate, "数据更新出错", "") } if userInfo.Id <= 0 { return nil, srvdef.New(srvdef.DBUpdate, "为0", "") } if score > 0 && onAdd != nil { onAdd(userInfo.Id, score) } return userInfo, nil } // UserPay 扣除用户充值金币 func UserPay(ctx context.Context, userId int64, amount int64) (*User, *srvdef.SrvErr) { if amount <= 0 { return nil, srvdef.New(srvdef.ErrParamError, "扣除金额不能小于0", "扣除金额不能小于0") } userInfo := &User{} err := db.FindOneAndUpdate(ctx, &userInfo, bson.M{ FN.User.Id: userId, FN.User.Recharge: bson.M{"$gte": amount}, }, bson.M{ "$inc": bson.M{ FN.User.Recharge: -amount, FN.User.ConsumptionTotal: amount, }, "$set": bson.M{FN.User.UpdatedAt: time.Now()}, }) if err != nil { return nil, srvdef.New(srvdef.DBUpdate, "数据更新出错", "") } if userInfo.Id <= 0 { return nil, srvdef.New(srvdef.DBUpdate, "余额不足或用户不存在", "") } return userInfo, nil } // FindOneUpdateAndReturnNew 更新用户数据并返回新的用户数据 func FindOneUpdateAndReturnNew(ctx context.Context, query bson.M, update bson.M, opts ...*options.FindOneAndUpdateOptions) (*User, *srvdef.SrvErr) { userInfo := &User{} err := db.FindOneAndUpdate(ctx, &userInfo, query, update) if err != nil { return nil, srvdef.New(srvdef.DBUpdate, "数据更新出错", "") } if userInfo.Id <= 0 { return nil, srvdef.New(srvdef.ErrUserNotExsit, "", "") } return userInfo, nil } // GetMapUsers 获取 map 结构的 users func GetMapUsers(ctx context.Context, ids []int64) (UserMap, *srvdef.SrvErr) { mapUsers := make(UserMap) if len(ids) == 0 { return mapUsers, nil } listUsers, serr := FindManyUser(ctx, bson.M{FN.User.Id: bson.M{"$in": ids}}) if serr != nil { return mapUsers, serr } for _, user := range listUsers { mapUsers[user.Id] = user } return mapUsers, nil } // FindManyUserByQuery 从 user 表中查找一批数据(自定义 bind) func FindManyUserByQuery(ctx context.Context, bind interface{}, query bson.M, opts ...*options.FindOptions) *srvdef.SrvErr { if err := db.Find(ctx, bind, query, opts...); err != nil { log.Warn("user:FindManyUserByQuery db.Find failed", log.E(err), log.Any("filter", query)) return srvdef.New(srvdef.ErrDbQueryError, "", "") } return nil } // InsertManyUser 插入一批用户 func InsertManyUser(ctx context.Context, users *UserList) *srvdef.SrvErr { if _, err := db.InsertMany(ctx, users); err != nil { log.Warn("User:InsertManyUser db.InsertMany failed", log.E(err)) return srvdef.New(srvdef.ErrDbInsertError, "", "") } return nil } // StatcenterSyncList 按创建时间增量拉取用户列表(数据统计同步用) func StatcenterSyncList(ctx context.Context, createAt time.Time, size int64) (UserList, error) { opt := (&options.FindOptions{}) opt.SetLimit(size) opt.SetSort(bson.M{FN.User.Id: 1}) userList := UserList{} err := db.Find(ctx, &userList, bson.M{FN.User.CreatedAt: bson.M{"$gt": createAt}}, opt) if err != nil { return nil, err } return userList, nil } // StatcenterSyncListByUserID 按 UserID 游标拉取注册用户(新统计中心) func StatcenterSyncListByUserID(ctx context.Context, userID int64, size int64) (UserList, error) { opt := (&options.FindOptions{}) opt.SetLimit(size) opt.SetSort(bson.D{{Key: FN.User.Id, Value: 1}}) filter := bson.M{ FN.User.Id: bson.M{"$gt": userID}, } userList := UserList{} err := db.Find(ctx, &userList, filter, opt) if err != nil { return nil, err } return userList, nil } // StatcenterSyncBindUserList 增量拉取已正式注册(绑定邮箱)用户 func StatcenterSyncBindUserList(ctx context.Context, bindAt time.Time, size int64) (UserList, error) { opt := (&options.FindOptions{}).SetLimit(size).SetSort(bson.M{FN.User.UpdatedAt: 1}) filter := bson.M{ FN.User.IsRegister: true, FN.User.Email: bson.M{"$exists": true, "$gt": ""}, FN.User.UpdatedAt: bson.M{"$gt": bindAt}, } userList := UserList{} if err := db.Find(ctx, &userList, filter, opt); err != nil { return nil, err } return userList, nil } // StatcenterSyncInviteList 增量拉取带上级邀请码的用户 func StatcenterSyncInviteList(ctx context.Context, inviteAt time.Time, size int64) (UserList, error) { opt := (&options.FindOptions{}).SetLimit(size).SetSort(bson.M{FN.User.CreatedAt: 1}) filter := bson.M{ FN.User.AgentInviteCode: bson.M{"$nin": []interface{}{"", nil}}, FN.User.CreatedAt: bson.M{"$gt": inviteAt}, } userList := UserList{} if err := db.Find(ctx, &userList, filter, opt); err != nil { return nil, err } return userList, nil }