cadmin/server/internal/admin/server/cron.go

package server

import (
	"gadmin/internal/admin/forms"
	"gadmin/internal/admin/service"
	"github.com/gin-gonic/gin"
	"github.com/robfig/cron/v3"
	"net/http/httptest"
	"sync"
	"time"
)

var cronTab *cron.Cron

func RegisterJob() {
	//cronTab.AddFunc("@every 300s", RefreshConsumptionStatistics) // 每300秒刷新一次消费统计缓存
	cronTab.AddFunc("@every 10s", CheckNotice) // 每10秒检查一次是否有需要推送的广播
	cronTab.AddFunc("@every 10s", CheckEmail)  // 每10秒检查一次是否有需要推送的邮件
}

func cronStart() {
	cronTab = cron.New()
	RegisterJob()
	cronTab.Start()
}

var (
	checkNoticeLock sync.Mutex
	checkEmailLock  sync.Mutex
)

func CheckEmail() {
	/*	ok := checkEmailLock.TryLock()
		if !ok {
			logrus.Info("CheckNotice进程正在运行，跳过...")
			return
		}
		defer checkEmailLock.Unlock()

		var (
			ctx   = context.TODO()
			q     = query.Use(config.DB).AdminEmail
			lists []*model.AdminEmail
			now   = time.Now().Unix()
			err   error
		)

		if err = q.WithContext(ctx).Where(q.Status.In(consts.EmailStatusWait)).Where(q.SendAt.Lte(now)).Order(q.ID.Desc()).Scan(&lists); err != nil {
			logrus.Warnf("CheckEmail Scan lists err:%+v", err)
			return
		}

		if len(lists) == 0 {
			return
		}

		for _, v := range lists {
			logrus.Infof("CheckEmail SendEmailToWorld data:%+v", utility.DumpToJSON(v))

			var q2 = query.Use(config.DB).AdminEmail
			err = service.AdminEmail.SendEmailToWorld(v)
			if err != nil {
				logrus.Warnf("CheckEmail 邮件发送失败：%+v", err)
				_, err2 := q2.WithContext(ctx).Where(q2.ID.Eq(v.ID)).Updates(model.AdminEmail{Status: consts.EmailStatusErr})
				if err2 != nil {
					logrus.Warnf("CheckEmail 邮件更新失败：%+v", err2)
				}
				continue
			}

			// 发送成功
			_, err = q2.WithContext(ctx).Where(q2.ID.Eq(v.ID)).Updates(model.AdminEmail{
				Status: consts.EmailStatusSent,
				SendAt: time.Now().Unix(),
			})
			if err != nil {
				logrus.Warnf("CheckEmail 邮件更新失败-2：%+v", err)
			}
		}*/
}

func CheckNotice() {
	/*	ok := checkNoticeLock.TryLock()
		if !ok {
			logrus.Info("CheckNotice进程正在运行，跳过...")
			return
		}
		defer checkNoticeLock.Unlock()
		var (
			ctx             = context.TODO()
			q               = query.Use(config.DB).AdminNotice
			lists           []*model.AdminNotice
			err             error
			normLabel       = "stopServerNotice"    // 普通广播
			stopServerLabel = "stopServerNoticeEnd" // 最后一次停服广播
			allServerIds    = service.ServerOption.GetServerIds()
		)

		err = q.WithContext(ctx).
			Where(q.Environment.Eq(os.Getenv("GIN_MODE"))).
			Where(q.Status.In(consts.NoticeStatusRunning, consts.NoticeStatusNotStarted)).
			Order(q.ID.Desc()).Scan(&lists)
		if err != nil {
			logrus.Warnf("CheckNotice Scan lists err:%+v", err)
			return
		}

		if len(lists) == 0 {
			//logrus.Info("CheckNotice 没有正在运行的广播..")
			return
		}

		updateStatus := func(models *model.AdminNotice) {
			_, err = query.Use(config.DB).AdminNotice.WithContext(ctx).
				Where(q.ID.Eq(models.ID)).
				Where(q.Environment.Eq(os.Getenv("GIN_MODE"))).
				Updates(&model.AdminNotice{LastSendAt: models.LastSendAt, Status: models.Status, UpdatedAt: time.Now().Unix()})
			if err != nil {
				logrus.Warnf("CheckNotice updateStatus err:%+v", err)
				return
			}
			return
		}

		sendToWorld := func(serverIds []int, msgId, typ, content string) {

			for _, sId := range serverIds {
				DB, err := player.GetDBByServerID(sId)
				if err != nil {
					logrus.Warnf("CheckNotice GetDBByServerID2 GetDBByServerID err:%+v", err)
					continue
				}

				msgItem := msg.GmPlacard{
					MsgId:   msgId,
					Type:    typ,
					Content: content,
				}
				//err = gate.SendToWorld(0, &msgItem)

				var resp *msg.ResponseGmPlacard
				res, err := config.GmNats.GmRequest(DB, "GmPlacard", msgItem)
				if err != nil {
					logrus.Warnf("CheckNotice GmRequest err:%+v", err)
					continue
				}

				if err = json.Unmarshal(res, &resp); err != nil {
					logrus.Warnf("CheckNotice nats Unmarshal err:%+v", err)
					continue
				}

				logrus.Infof("CheckNotice sendToWorld ok!! DB:%v, msg:%+v err:%+v", DB, msgItem, err)
			}

			return
		}

		for _, v := range lists {
			now := time.Now().Unix()

			var serverIds []int
			if v.ServerIds == "" {
				serverIds = allServerIds
			} else {
				if err := json.Unmarshal([]byte(v.ServerIds), &serverIds); err != nil {
					logrus.Warnf("CheckNotice Unmarshal err:%+v", err)
					continue
				}

				// 所有服务器
				for _, serverId := range serverIds {
					if serverId == 0 {
						serverIds = allServerIds
						break
					}
				}
			}

			// 未开始的广播
			if v.StartAt > now {
				continue
			}

			// 如果最后时间小于当前时间，不执行了
			if v.EndAt < now {
				v.Status = consts.NoticeStatusStopped
				// 关服广播
				if v.NoticeType == consts.NoticeTypeStopSev {
					sendToWorld(serverIds, v.MsgID, stopServerLabel, "")
				}
				updateStatus(v)
				continue
			}

			// 判断是否满足发送间隔
			if v.LastSendAt+v.SendInterval*60 < now {
				v.LastSendAt = now

				// 发送广播
				sendToWorld(serverIds, v.MsgID, normLabel, v.Content)

				if consts.NoticeStatusNotStarted == v.Status {
					v.Status = consts.NoticeStatusRunning
				}
				updateStatus(v)
				continue
			}
		}*/

}

func RefreshConsumptionStatistics() {
	var (
		ctx, _      = gin.CreateTestContext(httptest.NewRecorder())
		currentTime = time.Now()
		endTime     = time.Date(currentTime.Year(), currentTime.Month(), currentTime.Day(), 0, 0, 0, 0, currentTime.Location()).UnixMilli()
		startTime   = endTime - 86400000*7
		endDate     = time.UnixMilli(endTime).Format("2006-01-02")
		startDate   = time.UnixMilli(startTime).Format("2006-01-02")
		req         forms.ConsumptionStatisticsReq
	)

	req = forms.ConsumptionStatisticsReq{
		ListReq: forms.ListReq{Page: 1, PerPage: 15}, Type: 1, Source: 0,
		//Createtime: []int64{1662566400000, 1663171200000},
		//Date:       []string{"2022-09-08", "2022-09-15"},
	}

	req.Createtime = []int64{startTime, endTime}
	req.Date = []string{startDate, endDate}

	// 钻石
	service.ChangedLogs.Statistics(ctx, req)

	// 金币
	req.Type = 2
	service.ChangedLogs.Statistics(ctx, req)
}