Initial commit

service/consumer/handler.go

@@ -0,0 +1,991 @@
+package consumer
+
+import (
+	"context"
+	"fmt"
+	"stat/common/enum/consts"
+	"stat/common/log"
+	"stat/common/statutils"
+	"stat/common/topicmessage"
+	"stat/common/utils"
+	"stat/models/accessmod"
+	"stat/models/channelmod"
+	downloadinfo "stat/models/downloadinfo"
+	"stat/models/rechargegamemod"
+	"stat/models/rechargemod"
+	"stat/models/registermod"
+	"stat/models/syncgamerecharge"
+	"stat/models/syncgamewater"
+	"stat/models/syncrecharge"
+	"stat/models/syncrecord"
+	"stat/models/syncvip"
+	"stat/models/syncwithdraw"
+	"stat/models/withdrawGameOrdermod"
+	"stat/models/withdrawUserOrdermod"
+	"stat/statg"
+	"strings"
+
+	"time"
+
+	"github.com/pkg/errors"
+	"go.mongodb.org/mongo-driver/bson"
+	"go.uber.org/zap"
+)
+
+// 1.用户访问消息处理函数
+func UserAccessHandler(accessMsg *topicmessage.UserVisitMsg) error {
+
+	accessInfo := accessmod.AccessMod{
+		UserId:    accessMsg.UserId,
+		AppId:     accessMsg.AppId,
+		SysType:   accessMsg.SysType,
+		DevType:   accessMsg.DevType,
+		IP:        accessMsg.IP,
+		Version:   accessMsg.Version,
+		DevID:     accessMsg.DevID,
+		VisitAt:   accessMsg.VisitAt,
+		CreatedAt: time.Now(),
+	}
+
+	// 查找用户信息
+	userInfo, err := registermod.GetRegisterInfo(context.Background(), accessInfo.UserId, accessInfo.AppId)
+	if err != nil {
+		return err
+	}
+
+	accessInfo.IsDirect = userInfo.IsDirect
+	accessInfo.DistrictCode = userInfo.DistrictCode
+	accessInfo.RegisterTime = userInfo.RegisterTime
+	accessInfo.IsFormalUser = userInfo.IsFormal
+
+	if userInfo.UserId != accessInfo.UserId {
+		accessInfo.IsDirect = true
+	}
+
+	return accessmod.AddAccessLog(context.Background(), &accessInfo)
+}
+
+// 2.用户注册消息处理函数
+func UserRegisterHandler(registerMsg *topicmessage.UserRegisterMsg) error {
+	ctx := context.Background()
+
+	registerInfo := registermod.RegisterMod{
+		UserId:         registerMsg.UserId,
+		AppId:          registerMsg.AppId,
+		SysType:        registerMsg.SysType,
+		DevType:        registerMsg.DevType,
+		Mobile:         registerMsg.Mobile,
+		Name:           registerMsg.Name,
+		IP:             registerMsg.IP,
+		IsDirect:       true,
+		DistrictCode:   strings.TrimSpace(registerMsg.DistrictCode),
+		DC:             strings.TrimSpace(registerMsg.DistrictCode),
+		DistrictFrom:   registermod.DistrictFromNone,
+		PromCode:       registerMsg.PromCode,
+		RegisterTime:   registerMsg.RegisterTime,
+		ParentPromCode: registerMsg.PromSeqe,
+		UserAgent:      registerMsg.UserAgent,
+		IsDeduction:    false,
+	}
+	var ipChecked bool
+	// 检测
+	testUser, err := registermod.GetRegisterInfo(ctx, registerInfo.UserId, registerInfo.AppId)
+	if err != nil {
+		log.Error("UserRegisterHandler define err", log.Any("UserId", testUser.UserId), log.E(err))
+		return err
+	}
+	if testUser.UserId == registerInfo.UserId {
+		return nil // 已存在
+	}
+	if len(registerInfo.ParentPromCode) > 0 && len(registerInfo.DistrictCode) == 0 {
+		testUser, err := registermod.GetRegisterInfoByInviteCode(ctx, registerInfo.ParentPromCode, registerInfo.AppId)
+		if err != nil || testUser.PromCode != registerInfo.ParentPromCode {
+			log.Error("GetRegisterInfoByInviteCode define err", log.Any("UserId", testUser.UserId), log.E(err))
+		} else {
+			// 推广相关逻辑
+			registerInfo.IsDirect = false
+			registerInfo.DistrictFrom = registermod.DistrictFromPC
+			registerInfo.DistrictCode = testUser.DistrictCode
+		}
+	}
+
+	if registerInfo.IP != "" {
+		if registerInfo.DistrictCode != "" {
+			channelInfo, _ := channelmod.FindOneChannelInfo(ctx, bson.M{channelmod.FN.ChannelInfo.Channel: registerInfo.DistrictCode})
+			if channelInfo == nil || channelInfo.Channel != registerInfo.DistrictCode {
+				registerInfo.DistrictCode = ""
+			}
+		}
+
+		if registerInfo.DistrictCode == "" {
+			// 根据IP查询是否下载
+			ipChecked = true
+			data, _ := downloadinfo.FindOneChannelIP(ctx, registerInfo.AppId, registerInfo.IP, registerInfo.SysType)
+			if data.DC != "" {
+				log.Info("channel info by IP ", log.Any("IP", registerInfo.IP), log.Any("app", registerInfo.AppId), log.Any("dc", data.DC))
+				registerInfo.DistrictFrom = registermod.DistrictFromIP
+				registerInfo.IpMatched = true
+				registerInfo.DistrictCode = data.DC
+			}
+		}
+	}
+
+	if registerInfo.DistrictCode != "" && registerInfo.DistrictFrom == registermod.DistrictFromNone {
+		registerInfo.DistrictFrom = registermod.DistrictFromDC
+	}
+	if !ipChecked && registerInfo.IP != "" {
+		data, _ := downloadinfo.FindOneChannelIP(ctx, registerInfo.AppId, registerInfo.IP, registerInfo.SysType)
+		if data.DC != "" {
+			registerInfo.IpMatched = true
+		}
+	}
+	registerInfo.BlocDeduction = false // 不扣量
+	// 插入记录
+	return registermod.AddRegisterLog(ctx, &registerInfo)
+}
+
+// 2.用户注册消息处理函数
+func FormalUserRegisterHandler(registerMsg *topicmessage.UserRegisterMsg) error {
+	ctx := context.Background()
+
+	registerInfo := registermod.RegisterMod{
+		UserId:         registerMsg.UserId,
+		AppId:          registerMsg.AppId,
+		SysType:        registerMsg.SysType,
+		DevType:        registerMsg.DevType,
+		Mobile:         registerMsg.Mobile,
+		Name:           registerMsg.Name,
+		IP:             registerMsg.IP,
+		IsDirect:       true,
+		DistrictCode:   strings.TrimSpace(registerMsg.DistrictCode),
+		DC:             strings.TrimSpace(registerMsg.DistrictCode),
+		DistrictFrom:   registermod.DistrictFromNone,
+		PromCode:       registerMsg.PromCode,
+		IsFormal:       true,
+		FormalTime:     registerMsg.RegisterTime,
+		RegisterTime:   registerMsg.RegisterTime,
+		ParentPromCode: registerMsg.PromSeqe,
+		IsDeduction:    false,
+	}
+
+	// 检测
+	testUser, err := registermod.GetRegisterInfo(ctx, registerInfo.UserId, registerInfo.AppId)
+	if err != nil {
+		log.Error("FormalUserRegisterHandler define err", log.Any("UserId", testUser.UserId), log.E(err))
+		return err
+	}
+	if testUser.UserId == registerInfo.UserId {
+		if testUser.RegisterTime.Before(registerInfo.FormalTime) {
+			// 更新
+			setdoc := bson.M{"$set": bson.M{
+				"formalTime": registerInfo.FormalTime,
+				"isFormal":   true,
+			}}
+			filter := bson.M{
+				"appId":  registerInfo.AppId,
+				"userId": registerInfo.UserId,
+			}
+			err := registermod.UpdateOneRegister(ctx, filter, setdoc)
+			if err != nil {
+				log.Error("NewCountMethod UpdateManyRegister ", log.E(err))
+			}
+		} else {
+			log.Error("FormalUserRegisterHandler register or formal time err", log.Any("AppId", testUser.AppId), log.Any("UserId", testUser.UserId), log.Any("RegisterTime", testUser.RegisterTime), log.Any("FormalTime", registerInfo.FormalTime))
+			return nil
+		}
+	} else {
+		// 插入
+		if len(registerInfo.ParentPromCode) > 0 && len(registerInfo.DistrictCode) == 0 {
+			testUser, err := registermod.GetRegisterInfoByInviteCode(ctx, registerInfo.ParentPromCode, registerInfo.AppId)
+			if err != nil || testUser.PromCode != registerInfo.ParentPromCode {
+				log.Error("GetRegisterInfoByInviteCode define err", log.Any("UserId", testUser.UserId), log.E(err))
+			} else {
+				// 推广相关逻辑
+				registerInfo.IsDirect = false
+				registerInfo.DistrictFrom = registermod.DistrictFromPC
+				registerInfo.DistrictCode = testUser.DistrictCode
+			}
+		} else if registerInfo.DistrictCode == "" && registerInfo.IP != "" {
+			data, _ := downloadinfo.FindOneChannelIP(ctx, registerInfo.AppId, registerInfo.IP, registerInfo.SysType)
+			if data.DC != "" {
+				log.Info("channel info by IP ", log.Any("IP", registerInfo.IP), log.Any("app", registerInfo.AppId), log.Any("dc", data.DC))
+				registerInfo.DistrictFrom = registermod.DistrictFromIP
+				registerInfo.DistrictCode = data.DC
+			}
+		}
+
+		registerInfo.BlocDeduction = false // 不扣量
+
+		// 插入记录
+		return registermod.AddRegisterLog(ctx, &registerInfo)
+	}
+	return nil
+}
+
+// 3.用户充值消息处理函数
+func UserReChargeHandler(rechargeMsg *topicmessage.UserRechargeMsg) error {
+	ctx := context.Background()
+
+	rechargeInfo := rechargemod.RechargeMod{
+		UserId:      rechargeMsg.UserId,
+		AppId:       rechargeMsg.AppId,
+		PlatformId:  rechargeMsg.PlatformId,
+		SysType:     rechargeMsg.SysType,
+		DevType:     rechargeMsg.DevType,
+		ChannelName: rechargeMsg.ChannelName,
+		CID:         rechargeMsg.CID,
+		Type:        rechargeMsg.Type,
+		OrderId:     rechargeMsg.OrderId,
+		Money:       rechargeMsg.Money,
+		PayMoney:    rechargeMsg.PayMoney,
+		Status:      3,
+		Rate:        rechargeMsg.Rate,
+		SuccessAt:   rechargeMsg.SuccessAt,
+		Repeat:      false,
+	}
+	// 检测订单是否存在
+	testRechargere, err := rechargemod.FindRechargeByOrder(ctx, rechargeInfo.AppId, rechargeInfo.OrderId)
+	if err != nil || testRechargere.OrderId == rechargeInfo.OrderId {
+		log.Error("UserReChargeHandler define err ", log.Any("OrderId", testRechargere.OrderId), log.E(err))
+		return err
+	}
+	if testRechargere.OrderId == rechargeInfo.OrderId {
+		return nil
+	}
+
+	// 查找用户信息
+	userInfo, err := registermod.GetRegisterInfo(context.Background(), rechargeInfo.UserId, rechargeInfo.AppId)
+	if err != nil {
+		log.Error("UserReChargeHandler GetRegisterInfo failed", log.Any("UserId", rechargeInfo.UserId), log.Any("AppId", rechargeInfo.AppId), log.Any("UserId", userInfo.UserId), log.E(err))
+		return err
+	}
+	if userInfo.UserId == rechargeInfo.UserId {
+		rechargeInfo.IsDirect = userInfo.IsDirect
+		rechargeInfo.DistrictCode = userInfo.DistrictCode
+		rechargeInfo.RegisterTime = userInfo.RegisterTime
+		rechargeInfo.PromCode = userInfo.PromCode
+		rechargeInfo.IsFormalUser = userInfo.IsFormal
+		// 判断用户是否充值
+		hasRechargere, _ := rechargemod.FindRechargeByUser(ctx, rechargeInfo.AppId, int64(rechargeInfo.UserId))
+		if hasRechargere.UserId == rechargeInfo.UserId {
+			rechargeInfo.Repeat = true
+		}
+
+	}
+	// 更新记录
+	err = rechargemod.AddRechargeLog(ctx, &rechargeInfo)
+	if err != nil {
+		log.Error("UserRegisterHandler AddRechargeLog err", log.E(err))
+		return err
+	}
+
+	// 集团
+	return RechargeChannelHandle(ctx, &rechargeInfo)
+
+	//fmt.Println("DistrictCode kafka recharge queue:", rechargeInfo)
+}
+
+// 4.用户提现
+func UserWithdrawOrder(UserWithdrawMsg *topicmessage.UserWithdrawMsg) error {
+	ctx := context.Background()
+
+	withdrawInfo := withdrawUserOrdermod.UserWithdrawOrder{
+		AppId:      UserWithdrawMsg.AppId,
+		UID:        UserWithdrawMsg.UID,
+		Name:       UserWithdrawMsg.Name,
+		OID:        UserWithdrawMsg.OID,
+		Money:      UserWithdrawMsg.Money,
+		PayMoney:   UserWithdrawMsg.PayMoney,
+		PayType:    UserWithdrawMsg.PayType,
+		ActName:    UserWithdrawMsg.ActName,
+		Act:        UserWithdrawMsg.Act,
+		DeviceType: UserWithdrawMsg.DeviceType,
+		DevID:      UserWithdrawMsg.DevID,
+		Status:     UserWithdrawMsg.Status,
+		SuccessAt:  UserWithdrawMsg.SuccessAt,
+		UpdatedAt:  UserWithdrawMsg.UpdatedAt,
+		CreatedAt:  UserWithdrawMsg.CreatedAt,
+	}
+
+	userInfo, err := registermod.GetRegisterInfo(ctx, UserWithdrawMsg.UID, UserWithdrawMsg.AppId)
+	if err != nil {
+		log.Error("rechargeDeduction GetRegisterInfo failed", log.Any("UserId", UserWithdrawMsg.UID), log.Any("AppId", UserWithdrawMsg.UID), log.Any("UserId", userInfo.UserId), log.E(err))
+		return err
+	}
+
+	// 兼容找不到用户
+	if userInfo.UserId == UserWithdrawMsg.UID {
+		withdrawInfo.IsDirect = userInfo.IsDirect
+		withdrawInfo.DistrictCode = userInfo.DistrictCode
+		withdrawInfo.PromCode = userInfo.PromCode
+		withdrawInfo.IsFormalUser = userInfo.IsFormal
+	} else {
+		withdrawInfo.IsDirect = true
+	}
+	// 更新记录
+	_, err = withdrawUserOrdermod.AddWithdrawOrder(ctx, &withdrawInfo)
+	if err != nil {
+		log.Error("UserWithdrawOrder  AddWithdrawOrder: %s\n", log.E(err))
+		return err
+	}
+	return nil
+	//fmt.Println("DistrictCode kafka withdraw queue:", withdrawInfo)
+}
+
+// 4.充值全量
+func UserSyncRechargeOrderALL(rechargeOrder *topicmessage.UserRechargeMsg) error {
+	ctx := context.Background()
+	rechargeInfo := &syncrecharge.SyncRechargeOrder{
+		UserId:      rechargeOrder.UserId,
+		AppId:       rechargeOrder.AppId,
+		PlatformId:  rechargeOrder.PlatformId,
+		SysType:     rechargeOrder.SysType,
+		DevType:     rechargeOrder.DevType,
+		ChannelName: rechargeOrder.ChannelName,
+		CID:         rechargeOrder.CID,
+		Type:        rechargeOrder.Type,
+		OrderId:     rechargeOrder.OrderId,
+		Money:       rechargeOrder.Money,
+		PayMoney:    rechargeOrder.PayMoney,
+		Status:      rechargeOrder.Status,
+		Rate:        rechargeOrder.Rate,
+		SuccessAt:   rechargeOrder.SuccessAt,
+		CreatedAt:   rechargeOrder.CreatedAt,
+	}
+	userInfo, err := registermod.GetRegisterInfo(ctx, rechargeOrder.UserId, rechargeOrder.AppId)
+	if err != nil {
+		return err
+	}
+	if userInfo.UserId == rechargeInfo.UserId {
+		rechargeInfo.IsDirect = userInfo.IsDirect
+		rechargeInfo.DistrictCode = userInfo.DistrictCode
+		rechargeInfo.PromCode = userInfo.PromCode
+		rechargeInfo.IsFormalUser = userInfo.IsFormal
+	} else {
+		rechargeInfo.IsDirect = true
+	}
+	if rechargeInfo.AppId == 494 {
+		log.Error("InsertOneSyncRechargeOrder@@@", zap.Any("rechargeInfo", rechargeInfo))
+		orderinfo, _ := syncrecharge.FindOneSyncRechargeOrder(ctx, bson.M{syncrecharge.FN.SyncRechargeOrder.AppId: rechargeInfo.AppId,
+			syncrecharge.FN.SyncRechargeOrder.UserId: rechargeInfo.UserId, syncrecharge.FN.SyncRechargeOrder.PlatformId: rechargeInfo.PlatformId})
+		if rechargeInfo.PlatformId == orderinfo.PlatformId {
+			log.Error("InsertOneSyncVip repeat", zap.Any("rechargeInfo", rechargeInfo))
+			return nil
+		}
+	}
+	return syncrecharge.InsertOneSyncRechargeOrder(ctx, rechargeInfo)
+}
+
+// 4.提现全量
+func UserSyncWithdrawOrderALL(UserWithdrawMsg *topicmessage.UserWithdrawMsg) error {
+	ctx := context.Background()
+	withdrawOrder := &syncwithdraw.SyncWithdraw{
+		AppId:      UserWithdrawMsg.AppId,
+		UID:        UserWithdrawMsg.UID,
+		Name:       UserWithdrawMsg.Name,
+		OID:        UserWithdrawMsg.OID,
+		Money:      UserWithdrawMsg.Money,
+		PayMoney:   UserWithdrawMsg.PayMoney,
+		PayType:    UserWithdrawMsg.PayType,
+		ActName:    UserWithdrawMsg.ActName,
+		Act:        UserWithdrawMsg.Act,
+		DeviceType: UserWithdrawMsg.DeviceType,
+		DevID:      UserWithdrawMsg.DevID,
+		Status:     UserWithdrawMsg.Status,
+		SuccessAt:  UserWithdrawMsg.SuccessAt,
+		UpdatedAt:  UserWithdrawMsg.UpdatedAt,
+		CreatedAt:  UserWithdrawMsg.CreatedAt,
+	}
+	userInfo, err := registermod.GetRegisterInfo(ctx, withdrawOrder.UID, withdrawOrder.AppId)
+	if err != nil {
+		log.Error("UserSyncWithdrawOrderALL GetRegisterInfo failed", log.Any("UserId", withdrawOrder.UID), log.Any("AppId", withdrawOrder.AppId), log.Any("UserId", userInfo.UserId), log.E(err))
+		return err
+	}
+	if userInfo.UserId != withdrawOrder.UID {
+		log.Error("UserSyncWithdrawOrderALL GetRegisterInfo failed", log.Any("UserId", withdrawOrder.UID), log.Any("AppId", withdrawOrder.AppId), log.Any("UserId", userInfo.UserId), log.E(err))
+	}
+	withdrawOrder.IsDirect = userInfo.IsDirect
+	withdrawOrder.DistrictCode = userInfo.DistrictCode
+	withdrawOrder.PromCode = userInfo.PromCode
+	withdrawOrder.IsFormalUser = userInfo.IsFormal
+
+	err = syncwithdraw.InsertOneSyncWithdraw(ctx, withdrawOrder)
+	if err != nil {
+		log.Error("UserSyncWithdrawOrderALL InsertOneSyncWithdraw failed", log.Any("UserId", withdrawOrder.UID), log.Any("AppId", withdrawOrder.AppId), log.Any("UserId", userInfo.UserId), log.E(err))
+	}
+	return err
+}
+
+// 5.VIP相关处理
+func UserSyncVIPOrderALL(UserBuyVipwMsg *topicmessage.UserBuyVIPMsg) error {
+	ctx := context.Background()
+	buyVipOrder := &syncvip.SyncVip{
+		PlatformId:  UserBuyVipwMsg.PlatformId,
+		AppId:       UserBuyVipwMsg.AppId,
+		UID:         UserBuyVipwMsg.UID,
+		Name:        UserBuyVipwMsg.Name,
+		Money:       UserBuyVipwMsg.Money,        // 单位转换成分
+		Amount:      UserBuyVipwMsg.Amount * 100, // 单位转换成分
+		BuyVIPType:  UserBuyVipwMsg.BuyVIPType,
+		OID:         UserBuyVipwMsg.OID,
+		DeviceType:  UserBuyVipwMsg.DeviceType,
+		Status:      UserBuyVipwMsg.Status,
+		Repeat:      false,
+		SuccessAt:   UserBuyVipwMsg.SuccessAt,
+		IsDeduction: false,
+	}
+	userInfo, err := registermod.GetRegisterInfo(ctx, UserBuyVipwMsg.UID, UserBuyVipwMsg.AppId)
+	if err != nil {
+		log.Error("UserSyncVIPOrderALL GetRegisterInfo failed", log.Any("UserId", UserBuyVipwMsg.UID), log.Any("AppId", UserBuyVipwMsg.AppId), log.Any("UserId", userInfo.UserId), log.E(err))
+		return err
+	}
+	if userInfo.UserId == buyVipOrder.UID {
+		buyVipOrder.IsDirect = userInfo.IsDirect
+		buyVipOrder.DistrictCode = userInfo.DistrictCode
+		buyVipOrder.PromCode = userInfo.PromCode
+		buyVipOrder.RegisterTime = userInfo.RegisterTime
+		buyVipOrder.IsFormalUser = userInfo.IsFormal
+
+		orderinfo, _ := syncvip.FindOneSyncVip(ctx, bson.M{syncvip.FN.SyncVip.AppId: buyVipOrder.AppId, syncvip.FN.SyncVip.UID: buyVipOrder.UID})
+		if orderinfo.UID == buyVipOrder.UID {
+			buyVipOrder.Repeat = true
+		}
+	}
+	if buyVipOrder.AppId == 494 {
+		log.Error("InsertOneSyncVip@@@", zap.Any("buyVipOrder", buyVipOrder))
+		orderinfo, _ := syncvip.FindOneSyncVip(ctx, bson.M{syncvip.FN.SyncVip.AppId: buyVipOrder.AppId,
+			syncvip.FN.SyncVip.UID: buyVipOrder.UID, syncvip.FN.SyncVip.PlatformId: buyVipOrder.PlatformId})
+		if orderinfo.PlatformId == buyVipOrder.PlatformId {
+			log.Error("InsertOneSyncVip repeat", zap.Any("buyVipOrder", buyVipOrder))
+			return nil
+		}
+	}
+	err = syncvip.InsertOneSyncVip(ctx, buyVipOrder)
+	if err != nil {
+		log.Error("UserSyncVIPOrderALL InsertOneSyncVip failed", log.Any("UserId", UserBuyVipwMsg.UID), log.Any("AppId", UserBuyVipwMsg.AppId), log.Any("UserId", userInfo.UserId), log.E(err))
+	}
+	return err
+}
+
+// 7.游戏充值
+func GameReChargeHandler(rechargeMsg *topicmessage.UserRechargeMsg) error {
+	ctx := context.Background()
+
+	rechargeInfo := rechargegamemod.GameRechargeMod{
+		UserId:      rechargeMsg.UserId,
+		AppId:       rechargeMsg.AppId,
+		PlatformId:  rechargeMsg.PlatformId,
+		SysType:     rechargeMsg.SysType,
+		DevType:     rechargeMsg.DevType,
+		ChannelName: rechargeMsg.ChannelName,
+		CID:         rechargeMsg.CID,
+		Type:        rechargeMsg.Type,
+		OrderId:     rechargeMsg.OrderId,
+		Money:       rechargeMsg.Money,
+		PayMoney:    rechargeMsg.PayMoney,
+		Status:      3,
+		SyncStatus:  rechargeMsg.SyncStatus,
+		Rate:        rechargeMsg.Rate,
+		SuccessAt:   rechargeMsg.SuccessAt,
+		Repeat:      false,
+	}
+	// 检测订单是否存在
+	testRechargere, err := rechargegamemod.FindRechargeByOrder(ctx, rechargeInfo.AppId, rechargeInfo.OrderId)
+	if err != nil {
+		return err
+	}
+	if testRechargere.OrderId == rechargeInfo.OrderId {
+		return nil // 订单已存在
+	}
+
+	// 查找用户信息
+	userInfo, err := registermod.GetRegisterInfo(context.Background(), rechargeInfo.UserId, rechargeInfo.AppId)
+	if err != nil {
+		log.Error("UserAccessHandler GetRegisterInfo failed", log.Any("UserId", rechargeInfo.UserId), log.Any("AppId", rechargeInfo.AppId), log.Any("UserId", userInfo.UserId), log.E(err))
+		return err
+	}
+	if userInfo.UserId == rechargeInfo.UserId {
+		rechargeInfo.IsDirect = userInfo.IsDirect
+		rechargeInfo.DistrictCode = userInfo.DistrictCode
+		rechargeInfo.RegisterTime = userInfo.RegisterTime
+		rechargeInfo.PromCode = userInfo.PromCode
+		rechargeInfo.IsFormalUser = userInfo.IsFormal
+		// 判断用户是否充值
+		hasRechargere, _ := rechargegamemod.FindRechargeByUser(ctx, rechargeInfo.AppId, int64(rechargeInfo.UserId))
+		if hasRechargere.UserId == rechargeInfo.UserId {
+			rechargeInfo.Repeat = true
+		}
+	}
+	// 更新记录
+	err = rechargegamemod.AddRechargeLog(ctx, &rechargeInfo)
+	if err != nil {
+		log.Error("UserRegisterHandler AddRechargeLog err", log.E(err))
+		return err
+	}
+
+	return nil
+	// 集团
+	// RechargeChannelHandle(ctx, &rechargeInfo)
+
+	//fmt.Println("DistrictCode kafka recharge queue:", rechargeInfo)
+
+}
+
+// 8.游戏充值全量
+func GameRechargeOrderALL(rechargeOrder *topicmessage.UserRechargeMsg) error {
+	ctx := context.Background()
+	rechargeInfo := &syncgamerecharge.SyncGameRechargeOrder{
+		UserId:      rechargeOrder.UserId,
+		AppId:       rechargeOrder.AppId,
+		PlatformId:  rechargeOrder.PlatformId,
+		SysType:     rechargeOrder.SysType,
+		DevType:     rechargeOrder.DevType,
+		ChannelName: rechargeOrder.ChannelName,
+		CID:         rechargeOrder.CID,
+		Type:        rechargeOrder.Type,
+		OrderId:     rechargeOrder.OrderId,
+		Money:       rechargeOrder.Money,
+		PayMoney:    rechargeOrder.PayMoney,
+		Status:      rechargeOrder.Status,
+		Rate:        rechargeOrder.Rate,
+		SyncStatus:  rechargeOrder.SyncStatus,
+		SuccessAt:   rechargeOrder.SuccessAt,
+		CreatedAt:   rechargeOrder.CreatedAt,
+	}
+	userInfo, err := registermod.GetRegisterInfo(ctx, rechargeOrder.UserId, rechargeOrder.AppId)
+	if err != nil {
+		return err
+	}
+
+	if userInfo.UserId == rechargeInfo.UserId {
+		rechargeInfo.IsDirect = userInfo.IsDirect
+		rechargeInfo.DistrictCode = userInfo.DistrictCode
+		rechargeInfo.PromCode = userInfo.PromCode
+		rechargeInfo.IsFormalUser = userInfo.IsFormal
+	} else {
+		rechargeInfo.IsDirect = true
+	}
+
+	err = syncgamerecharge.InsertOneSyncGameRechargeOrder(ctx, rechargeInfo)
+	if err != nil {
+		log.Error("GameRechargeOrderALL InsertOneSyncRechargeOrder failed", log.Any("UserId", rechargeOrder.UserId), log.Any("AppId", rechargeOrder.AppId), log.Any("UserId", userInfo.UserId), log.E(err))
+	}
+	return err
+}
+
+// 9.游戏提现
+func GameWithdrawOrder(UserWithdrawMsg *topicmessage.UserWithdrawMsg) error {
+	ctx := context.Background()
+
+	withdrawInfo := withdrawGameOrdermod.GameWithdrawOrder{
+		AppId:      UserWithdrawMsg.AppId,
+		UID:        UserWithdrawMsg.UID,
+		Name:       UserWithdrawMsg.Name,
+		OID:        UserWithdrawMsg.OID,
+		Money:      UserWithdrawMsg.Money,
+		PayMoney:   UserWithdrawMsg.PayMoney,
+		PayType:    UserWithdrawMsg.PayType,
+		ActName:    UserWithdrawMsg.ActName,
+		Act:        UserWithdrawMsg.Act,
+		DeviceType: UserWithdrawMsg.DeviceType,
+		DevID:      UserWithdrawMsg.DevID,
+		Status:     UserWithdrawMsg.Status,
+		SuccessAt:  UserWithdrawMsg.SuccessAt,
+		SyncStatus: UserWithdrawMsg.SyncStatus,
+		PayChannel: UserWithdrawMsg.PayChannel,
+		UpdatedAt:  UserWithdrawMsg.UpdatedAt,
+		CreatedAt:  UserWithdrawMsg.CreatedAt,
+	}
+
+	userInfo, err := registermod.GetRegisterInfo(ctx, UserWithdrawMsg.UID, UserWithdrawMsg.AppId)
+	if err != nil {
+		log.Error("rechargeDeduction GetRegisterInfo failed", log.Any("UserId", UserWithdrawMsg.UID), log.Any("AppId", UserWithdrawMsg.UID), log.Any("UserId", userInfo.UserId), log.E(err))
+		return err
+	}
+
+	// 兼容找不到用户
+	if userInfo.UserId == UserWithdrawMsg.UID {
+		withdrawInfo.IsDirect = userInfo.IsDirect
+		withdrawInfo.DistrictCode = userInfo.DistrictCode
+		withdrawInfo.PromCode = userInfo.PromCode
+		withdrawInfo.IsFormalUser = userInfo.IsFormal
+	} else {
+		withdrawInfo.IsDirect = true
+	}
+
+	// 更新记录
+	_, err = withdrawGameOrdermod.AddWithdrawOrder(ctx, &withdrawInfo)
+	if err != nil {
+		log.Error("UserWithdrawOrder  AddWithdrawOrder: %s\n", log.E(err))
+	}
+
+	return err
+	//fmt.Println("DistrictCode kafka withdraw queue:", withdrawInfo)
+}
+
+// 10.游戏提现
+func UserGameWater(useGamerWater *topicmessage.UserGameWaterMsg) error {
+	ctx := context.Background()
+	now := time.Now()
+
+	gameWaterInfo := syncgamewater.SyncGameWater{
+		AppId:      useGamerWater.AppId,
+		UserId:     useGamerWater.UID,
+		PlatformId: useGamerWater.PlatformId,
+		Date:       useGamerWater.Date,
+		ValidBet:   useGamerWater.ValidBet,
+		Profit:     useGamerWater.Profit,
+		Revenue:    useGamerWater.Revenue,
+		CreatedAt:  now,
+	}
+
+	userInfo, err := registermod.GetRegisterInfo(ctx, useGamerWater.UID, useGamerWater.AppId)
+	if err != nil {
+		log.Error("rechargeDeduction GetRegisterInfo failed", log.Any("UserId", useGamerWater.UID), log.Any("AppId", useGamerWater.UID), log.Any("UserId", userInfo.UserId), log.E(err))
+		return err
+	}
+
+	// 兼容找不到用户
+	if userInfo.UserId == useGamerWater.UID {
+		gameWaterInfo.IsDirect = userInfo.IsDirect
+		gameWaterInfo.DistrictCode = userInfo.DistrictCode
+		gameWaterInfo.PromCode = userInfo.PromCode
+		gameWaterInfo.RegisterTime = userInfo.RegisterTime
+		gameWaterInfo.IsFormalUser = userInfo.IsFormal
+	} else {
+		gameWaterInfo.IsDirect = true
+	}
+	// 更新记录
+	err = syncgamewater.InsertOneSyncGameWater(ctx, &gameWaterInfo)
+	if err != nil {
+		log.Error("UserWithdrawOrder  AddWithdrawOrder: %s\n", log.E(err))
+	}
+
+	return err
+	//fmt.Println("DistrictCode kafka gamewater queue:", gameWaterInfo)
+}
+
+func RunSyncJob(ctx context.Context) error {
+	recoredResult, err := syncrecord.LastRecordTimeMap(ctx) //第一次从默认记录时间开始统计
+	if err != nil {
+		log.Error("statdistrict RunStat lastRecordTime failed!", log.E(err))
+		return err
+	}
+	for _, record := range recoredResult {
+		log.Info("RunSyncJob start", log.Any("start", time.Now()),
+			log.Any("record", record.AppId), log.Any("job", record.Job))
+		if record.Disabled {
+			continue
+		}
+		if !statg.GlobalEndpoints.GetIndexEndpoints(statg.Conf.ServerIndex).ExistsByAppId(record.AppId) {
+			continue // 不属于这个服务器的APP
+		}
+		// 任务开始
+		statutils.GlobalSyncMonitor.SetData(record.AppId, record.Job, func(item *statutils.SyncItem) {
+			item.SetJobBegin(time.Now())
+			item.CurrentCount = 0
+			item.Error = nil
+		})
+		err := processJobItem(ctx, &record)
+		// 任务结束
+		statutils.GlobalSyncMonitor.SetData(record.AppId, record.Job, func(item *statutils.SyncItem) {
+			item.Error = err
+			item.SetJobEnd(time.Now())
+		})
+
+		if err != nil {
+			var cErr *ContinueError
+			if !errors.As(err, &cErr) {
+				log.Info("RunSyncJob end err", log.Any("err", err))
+				// return err
+			}
+		}
+		log.Info("RunSyncJob end", log.Any("end", time.Now()),
+			log.Any("record", record.AppId), log.Any("job", record.Job))
+	}
+	return nil
+}
+
+func processJobRequest(ctx context.Context, record *syncrecord.SyncRecord) (*topicmessage.StatcenterSyncResp, error) {
+	request := &topicmessage.StatcenterSyncReq{
+		Job:         record.Job,
+		UserId:      record.UserId,
+		PlatformId:  record.PlatformId,
+		SuccessTime: record.SuccessAt,
+		MaxSize:     5000,
+	}
+	url, err := AppCallbackUrl(record.AppId)
+	if err != nil {
+		return nil, err
+	}
+	apiUrl := url + "/api/web/statcenter/sync"
+	var resp *topicmessage.StatcenterSyncResp
+	if err = HttpRequest(ctx, apiUrl, &request, &resp); err != nil {
+		return nil, err
+	}
+	if resp.Code != 200 {
+		return nil, NewContinueError(fmt.Sprintf("http request code %d, url: %s", resp.Code, apiUrl))
+	}
+	return resp, nil
+}
+
+func processJobItem(ctx context.Context, record *syncrecord.SyncRecord) error {
+	// 请求开始
+	statutils.GlobalSyncMonitor.SetData(record.AppId, record.Job, func(item *statutils.SyncItem) {
+		item.RequestBegin = time.Now()
+	})
+	resp, err := processJobRequest(ctx, record)
+	if err != nil {
+		return err
+	}
+	// 请求结束
+	statutils.GlobalSyncMonitor.SetData(record.AppId, record.Job, func(item *statutils.SyncItem) {
+		item.Error = err
+		item.RequestEnd = time.Now()
+	})
+	if err != nil {
+		return err
+	}
+
+	// 处理开始
+	statutils.GlobalSyncMonitor.SetData(record.AppId, record.Job, func(item *statutils.SyncItem) {
+		item.ProcessBegin = time.Now()
+	})
+	err = processJobData(ctx, record, resp, func(dataLatest time.Time, incValue int) {
+		// 处理中
+		statutils.GlobalSyncMonitor.SetData(record.AppId, record.Job, func(item *statutils.SyncItem) {
+			item.IncCount(int64(incValue))
+			item.DataLatest = dataLatest
+		})
+	})
+	// 处理结束
+	statutils.GlobalSyncMonitor.SetData(record.AppId, record.Job, func(item *statutils.SyncItem) {
+		item.ProcessEnd = time.Now()
+		item.Error = err
+	})
+	return err
+}
+
+func processJobData(ctx context.Context, record *syncrecord.SyncRecord, resp *topicmessage.StatcenterSyncResp, onItem func(dataLatest time.Time, incValue int)) error {
+	if record.Job == syncrecord.VisitJob {
+		for _, v := range resp.VisitList {
+			if err := UserAccessHandler(v); err != nil {
+				return err
+			}
+			record.UserId = v.UserId
+			record.SuccessAt = v.VisitAt
+			record.PlatformId = v.PlatformId
+
+			if err := syncrecord.UpsertItemRecord(ctx, record.Job, record.AppId, record.UserId, record.PlatformId, record.SuccessAt); err != nil {
+				log.Error("RunSyncJob 更新同步记录!", log.E(err))
+				return err
+			}
+		}
+		onItem(record.SuccessAt, len(resp.VisitList))
+	} else if record.Job == syncrecord.RechargeJob {
+		for _, v := range resp.RechargeList {
+			if err := UserReChargeHandler(v); err != nil {
+				return err
+			}
+			record.UserId = v.UserId
+			record.SuccessAt = v.SuccessAt
+			record.PlatformId = v.PlatformId
+			if err := syncrecord.UpsertItemRecord(ctx, record.Job, record.AppId, record.UserId, record.PlatformId, record.SuccessAt); err != nil {
+				log.Error("RunSyncJob 更新同步记录!", log.E(err))
+				return err // keep loop, nil
+			}
+		}
+		onItem(record.SuccessAt, len(resp.RechargeList))
+	} else if record.Job == syncrecord.RegisterJob {
+
+		for _, v := range resp.RegisterList {
+			if err := UserRegisterHandler(v); err != nil {
+				return err
+			}
+
+			record.PlatformId = v.PlatformId
+			record.SuccessAt = v.RegisterTime
+			record.UserId = v.UserId
+
+			if err := syncrecord.UpsertItemRecord(ctx, record.Job, record.AppId, record.UserId, record.PlatformId, record.SuccessAt); err != nil {
+				log.Error("RunSyncJob 更新同步记录!", log.E(err))
+				return err // keep loop, nil
+			}
+		}
+		onItem(record.SuccessAt, len(resp.RegisterList))
+	} else if record.Job == syncrecord.RegisterLightJob {
+		var InvalidDataCount int
+		var lastError error
+		for _, v := range resp.RegisterList {
+			if !statutils.IsClientH5OrLight(v.SysType, v.DevType) {
+				InvalidDataCount++
+				lastError = NewContinueError(fmt.Sprintf("系统类型不对,总无效:%d,Job:%s,App:%d,User:%d,SysType:%s,DevType:%s", InvalidDataCount, record.Job, record.AppId, v.UserId, v.SysType, v.DevType))
+				continue
+			}
+			if err := FormalUserRegisterHandler(v); err != nil {
+				return err
+			}
+
+			record.PlatformId = v.PlatformId
+			record.SuccessAt = v.RegisterTime
+			record.UserId = v.UserId
+
+			if err := syncrecord.UpsertItemRecord(ctx, record.Job, record.AppId, record.UserId, record.PlatformId, record.SuccessAt); err != nil {
+				log.Error("RunSyncJob 更新同步记录!", log.E(err))
+				return err // keep loop, nil
+			}
+		}
+		onItem(record.SuccessAt, len(resp.RegisterList)-InvalidDataCount)
+		if lastError != nil {
+			return lastError
+		}
+	} else if record.Job == syncrecord.WithdrawJob {
+		for _, v := range resp.WithdrawList {
+			if err := UserWithdrawOrder(v); err != nil {
+				return err
+			}
+
+			record.UserId = v.UID
+			record.SuccessAt = v.SuccessAt
+
+			if err := syncrecord.UpsertItemRecord(ctx, record.Job, record.AppId, record.UserId, record.PlatformId, record.SuccessAt); err != nil {
+				log.Error("RunSyncJob 更新同步记录!", log.E(err))
+				return err // keep loop, nil
+			}
+		}
+		onItem(record.SuccessAt, len(resp.WithdrawList))
+	} else if record.Job == syncrecord.VipJob {
+		if record.AppId == 494 {
+			log.Error("debug@@@", zap.Any("resp", resp))
+		}
+		for _, v := range resp.VipList {
+			if err := UserSyncVIPOrderALL(v); err != nil {
+				return err
+			}
+			record.UserId = v.UID
+			record.SuccessAt = v.SuccessAt
+			if err := syncrecord.UpsertItemRecord(ctx, record.Job, record.AppId, record.UserId, record.PlatformId, record.SuccessAt); err != nil {
+				log.Error("RunSyncJob 更新同步记录!", log.E(err))
+				return err // keep loop, nil
+			}
+		}
+		onItem(record.SuccessAt, len(resp.VipList))
+	} else if record.Job == syncrecord.WithdrawAll {
+		for _, v := range resp.WithdrawList {
+			if err := UserSyncWithdrawOrderALL(v); err != nil {
+				return err
+			}
+			record.UserId = v.UID
+			record.SuccessAt = v.CreatedAt
+			if err := syncrecord.UpsertItemRecord(ctx, record.Job, record.AppId, record.UserId, record.PlatformId, record.SuccessAt); err != nil {
+				log.Error("RunSyncJob 更新同步记录!", log.E(err))
+				return err // keep loop, nil
+			}
+		}
+		onItem(record.SuccessAt, len(resp.WithdrawList))
+	} else if record.Job == syncrecord.RechargeAll {
+		if record.AppId == 494 {
+			log.Error("debug@@@", zap.Any("resp", resp))
+		}
+		for _, v := range resp.RechargeList {
+			if err := UserSyncRechargeOrderALL(v); err != nil {
+				return err
+			}
+			record.UserId = v.UserId
+			record.SuccessAt = v.CreatedAt
+			if err := syncrecord.UpsertItemRecord(ctx, record.Job, record.AppId, record.UserId, record.PlatformId, record.SuccessAt); err != nil {
+				log.Error("RunSyncJob 更新同步记录!", log.E(err))
+				return err // keep loop, nil
+			}
+		}
+		onItem(record.SuccessAt, len(resp.RechargeList))
+	} else if record.Job == syncrecord.GameRechargeOrder {
+		for _, v := range resp.GameRecharge {
+			if err := GameReChargeHandler(v); err != nil {
+				return err
+			}
+			record.UserId = v.UserId
+			record.SuccessAt = v.SuccessAt
+			record.PlatformId = v.PlatformId
+			if err := syncrecord.UpsertItemRecord(ctx, record.Job, record.AppId, record.UserId, record.PlatformId, record.SuccessAt); err != nil {
+				log.Error("RunSyncJob 更新同步记录!", log.E(err))
+				return err // keep loop, nil
+			}
+		}
+		onItem(record.SuccessAt, len(resp.GameRecharge))
+	} else if record.Job == syncrecord.GameRechargeOrderAll {
+		for _, v := range resp.GameRecharge {
+			if err := GameRechargeOrderALL(v); err != nil {
+				return err
+			}
+			record.UserId = v.UserId
+			record.SuccessAt = v.CreatedAt
+			if err := syncrecord.UpsertItemRecord(ctx, record.Job, record.AppId, record.UserId, record.PlatformId, record.SuccessAt); err != nil {
+				log.Error("RunSyncJob 更新同步记录!", log.E(err))
+				return err // keep loop, nil
+			}
+		}
+		onItem(record.SuccessAt, len(resp.GameRecharge))
+	} else if record.Job == syncrecord.GameWithdrawOrder {
+		for _, v := range resp.GameWithdraw {
+			if err := GameWithdrawOrder(v); err != nil {
+				return err
+			}
+			record.UserId = v.UID
+			record.SuccessAt = v.SuccessAt
+			if err := syncrecord.UpsertItemRecord(ctx, record.Job, record.AppId, record.UserId, record.PlatformId, record.SuccessAt); err != nil {
+				log.Error("RunSyncJob 更新同步记录!", log.E(err))
+				return err // keep loop, nil
+			}
+		}
+		onItem(record.SuccessAt, len(resp.GameWithdraw))
+	} else if record.Job == syncrecord.GameWater {
+		for _, v := range resp.GameWater {
+			if err := UserGameWater(v); err != nil {
+				return err
+			}
+			record.UserId = v.UID
+			record.SuccessAt = v.SuccessAt
+			if err := syncrecord.UpsertItemRecord(ctx, record.Job, record.AppId, record.UserId, record.PlatformId, record.SuccessAt); err != nil {
+				log.Error("RunSyncJob 更新同步记录!", log.E(err))
+				return err // keep loop, nil
+			}
+		}
+		onItem(record.SuccessAt, len(resp.GameWater))
+	} else {
+		log.Error("RunSyncJob 类型不支持!", log.Any("job", record.Job))
+		return errors.New("RunSyncJob 类型不支持")
+	}
+	return nil
+}
+
+// 周期执行
+func SyncInit() {
+	if statg.Conf.RunMod != consts.RunModRelease {
+		return
+	}
+	defer utils.RecoverError(nil)
+	ticker := time.NewTicker(time.Second * 20) // 20秒一次
+	ctx := context.Background()
+
+	for t := range ticker.C {
+		fmt.Println("Begin task", t)
+		// 定时去app拉去任务
+		if statg.Conf.KafkaServer {
+			statutils.ServerMon.RunWithName(func() error {
+				return RunSyncJob(ctx)
+			}, statutils.GetFuncName(RunSyncJob))
+		}
+	}
+}
+
+func AppCallbackUrl(appid int32) (url string, err error) {
+	info := statg.GlobalAppIdEndpoints[appid]
+	if info != nil {
+		if len(info.ServerUrl) == 0 {
+			return "", NewContinueError(fmt.Sprintf("appid %d is not config server url", appid))
+		}
+		return info.ServerUrl, nil
+	}
+	return "", NewContinueError("appid err")
+}

service/consumer/landingpage.go

@@ -0,0 +1,12 @@
+package consumer
+
+import (
+	"context"
+	downloadinfo "stat/models/downloadinfo"
+)
+
+func TrackDcAndPcByIp(ctx context.Context, appId int32, ip, sysType string) (dc, pc string) {
+	data, _ := downloadinfo.FindOneChannelIP(ctx, appId, ip, sysType)
+
+	return data.DC, data.PC
+}

service/consumer/rechargechannel.go

@@ -0,0 +1,32 @@
+package consumer
+
+import (
+	"context"
+	"stat/common/log"
+	"stat/common/utils"
+	"stat/models/rechargechannelmod"
+	"stat/models/rechargemod"
+	"strconv"
+)
+
+func RechargeChannelHandle(ctx context.Context, rechageInfo *rechargemod.RechargeMod) error {
+	sumDate := utils.BeginWithTime(rechageInfo.SuccessAt)
+	rate := float64(4)
+	if rechageInfo.Rate != "" && rechageInfo.ChannelName != "daichong" {
+		rateF, err := strconv.ParseFloat(rechageInfo.Rate, 64)
+		if err != nil {
+			log.Error("RechargeChannelHandle rate is err", log.Any("rate", rechageInfo.Rate), log.Any("userid", rechageInfo.UserId), log.Any("appid", rechageInfo.AppId))
+			return err
+		}
+		rate = rateF
+	} else if rechageInfo.ChannelName == "daichong" {
+		rate = float64(8)
+		rechageInfo.CID = "daichong"
+	}
+	channelMoney := int64((rate * float64(rechageInfo.PayMoney)) / 100)
+	err := rechargechannelmod.UpsertOneUserRechargeStat(ctx, rechageInfo.CID, rechageInfo.AppId, sumDate, rechageInfo.BlocDeduction, rechageInfo.Money, rechageInfo.PayMoney, channelMoney)
+	if err != nil {
+		log.Error("RechargeChannelHandle UpsertOneUserRechargeStat", log.Any("rate", rechageInfo.Rate), log.Any("userid", rechageInfo.UserId), log.Any("appid", rechageInfo.AppId), log.E(err))
+	}
+	return err
+}

service/consumer/utils.go

@@ -0,0 +1,26 @@
+package consumer
+
+import (
+	"context"
+	"stat/common/utils"
+)
+
+type ContinueError struct {
+	msg string
+}
+
+func NewContinueError(text string) error {
+	ret := &ContinueError{
+		msg: text,
+	}
+	return ret
+}
+
+func (e *ContinueError) Error() string {
+	return e.msg
+}
+
+// HttpRequest HttpRequest
+func HttpRequest(ctx context.Context, url string, request interface{}, resp interface{}) error {
+	return utils.HttpPostRequestJson(url, request, resp)
+}