stat_center/service/consumer/handler.go

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")
}