cadmin/server/internal/cron/jobs/level_output.go

package jobs

import (
	"fmt"
	"gadmin/config"
	"gadmin/internal/gorm/model"
	"gadmin/internal/gorm/query"
	"gadmin/utility/player"
	"os"
	"sync"

	"github.com/sirupsen/logrus"
	"gorm.io/gen"
)

var LevelOutput = new(jLevelOutput)

type jLevelOutput struct {
	sync.Mutex
}

func (j *jLevelOutput) Run() {
	logrus.Info("LevelOutput Run.....")

	if os.Getenv("GIN_MODE") != "release" && os.Getenv("ADMIN_IS_LOCAL") != "1" {
		logrus.Warnf("测试环境禁止同步")
		return
	}

	j.Lock()
	defer j.Unlock()
	for serverId, _ := range config.GDBGroup {
		j.sync(serverId)
	}
}

func (j *jLevelOutput) sync(serverId int) {
	DB, err := player.GetDBByServerID(serverId)
	if err != nil {
		logrus.Warningf("SyncChannel GetDBByServerID err：%v", err)
		return
	}

	var (
		key       = fmt.Sprintf("level_output_sync_changed_id_%v", serverId)
		lastId    = config.FirstOrCreate(key, "sync")
		newLastId = lastId
		c         = query.Use(config.DB).ChangedLog
		results   []*model.ChangedLog
	)

	err = c.Select(c.ID, c.Diamond, c.Coin, c.UserID, c.ChannelID).Where(c.ID.Gt(lastId)).Where(c.ServerID.Eq(int32(serverId))).FindInBatches(&results, 500, func(tx gen.Dao, batch int) error {
		for _, result := range results {
			newLastId = result.ID

			if result.Coin <= 0 && result.Diamond <= 0 {
				continue
			}

			//logrus.Warnf("result:%+v", result)

			level := j.getLevel(DB, result.UserID)
			if level == 0 {
				//logrus.Warnf("无等级...")
				continue
			}

			if result.Coin > 0 {
				j.updateOutput(serverId, level, result.ChannelID, "coin", result.Coin)
			}

			if result.Diamond > 0 {
				j.updateOutput(serverId, level, result.ChannelID, "diamond", result.Diamond)
			}

		}
		return nil
	})

	if err != nil {
		logrus.Warnf("LevelOutput sync err:%+v", err)
		return
	}

	if newLastId != lastId {
		config.Save(key, newLastId)
	}

	logrus.Info("LevelOutput sync success..")
}

func (j *jLevelOutput) getLevel(DB int, userId int64) int32 {
	var models *model.PlayerAttr
	config.GDBGroup[DB].Select("level").Where("`playerid` = ?", userId).First(&models)
	if models == nil {
		return 0
	}

	return models.Level
}

func (j *jLevelOutput) updateOutput(serverId int, level int32, channelId, key string, incVal int64) {
	var (
		l   = query.Use(config.DB).ReportLevelOutput
		err error
	)

	first, err := l.Where(l.Level.Eq(int64(level)), l.ChannelID.Eq(channelId), l.ServerID.Eq(int32(serverId))).First()
	if err != nil && err.Error() != "record not found" {
		logrus.Errorf("updateOutput First err :%+v", err)
		return
	}

	// 新增
	if first == nil {
		data := &model.ReportLevelOutput{
			ID:        0,
			ServerID:  int32(serverId),
			Level:     int64(level),
			ChannelID: channelId,
			Coin:      0,
			Diamond:   0,
		}

		switch key {
		case "coin":
			data.Coin = incVal
		case "diamond":
			data.Diamond = incVal
		default:
			logrus.Warnf("updateOutput 新增 switch key err:%v", key)
		}
		config.DB.Create(data)
		return
	}

	// 更新
	switch key {
	case "coin":
		_, err = l.Where(l.ID.Eq(first.ID)).UpdateSimple(l.Coin.Add(incVal))
	case "diamond":
		_, err = l.Where(l.ID.Eq(first.ID)).UpdateSimple(l.Diamond.Add(incVal))
	default:
		logrus.Warnf("updateOutput 更新 switch key err:%v", key)
	}

	if err != nil {
		logrus.Errorf("updateOutput UpdateSimple err :%+v", err)
		return
	}
}