iotlcm/server/internal/service/allocatedispatch/service.go

package allocatedispatch

import (
	"context"
	"strings"
	"time"

	"github.com/gogf/gf/v2/frame/g"
	"github.com/gogf/gf/v2/os/gtime"
	"git.5iot.com/iotlcm-public/iotlcm-app/internal/consts"
	"git.5iot.com/iotlcm-public/iotlcm-app/internal/dao"
	"git.5iot.com/iotlcm-public/iotlcm-app/internal/library/batchmetadata"
	"git.5iot.com/iotlcm-public/iotlcm-app/internal/library/rabbitmq"
	"git.5iot.com/iotlcm-public/iotlcm-app/internal/model/entity"
)

type message struct {
	BatchNo   string `json:"batchNo"`
	BatchType int    `json:"batchType"`
}

func Start(ctx context.Context) context.CancelFunc {
	ctx, cancel := context.WithCancel(ctx)
	if !g.Cfg().MustGet(ctx, "workerBatchAsync.enabled", false).Bool() || !g.Cfg().MustGet(ctx, "workerBatchAsync.allocateRecovery.enabled", true).Bool() {
		return cancel
	}
	interval := g.Cfg().MustGet(ctx, "workerBatchAsync.allocateRecovery.intervalSeconds", 60).Int()
	if interval < 30 {
		interval = 30
	}
	go func() {
		ticker := time.NewTicker(time.Duration(interval) * time.Second)
		defer ticker.Stop()
		var cursor int64
		for {
			select {
			case <-ctx.Done():
				return
			case <-ticker.C:
				next, err := recoverPage(ctx, cursor, publish)
				if err != nil {
					g.Log().Errorf(ctx, "划拨任务补投失败: %+v", err)
				} else {
					cursor = next
				}
			}
		}
	}()
	return cancel
}

func due(batch *entity.BatchMain, now time.Time, processingDelay time.Duration) bool {
	if batch == nil || batch.BatchType != consts.BatchTypeAllocate || batch.SourceChannel != batchmetadata.MiniProgram {
		return false
	}
	delay := time.Minute
	switch batch.BatchAsyncStatus {
	case consts.BatchAsyncStatusPending:
	case consts.BatchAsyncStatusProcessing:
		delay = processingDelay
	default:
		return false
	}
	timestamp := batch.UpdatedAt
	if batch.BatchAsyncStatus == consts.BatchAsyncStatusProcessing && batch.StartTime != nil && !batch.StartTime.IsZero() {
		timestamp = batch.StartTime
	}
	if timestamp == nil || timestamp.IsZero() {
		timestamp = batch.CreatedAt
	}
	if timestamp == nil || timestamp.IsZero() {
		return false
	}
	return !timestamp.Time.Add(delay).After(now)
}

func recoverPage(ctx context.Context, cursor int64, send func(context.Context, string, string) error) (int64, error) {
	if ctx.Err() != nil {
		return cursor, ctx.Err()
	}
	columns := dao.BatchMain.Columns()
	var parents []*entity.BatchMain
	err := dao.BatchMain.Ctx(ctx).Where(columns.Id+" > ?", cursor).
		Where(columns.BatchType, consts.BatchTypeAllocate).Where(columns.SourceChannel, batchmetadata.MiniProgram).
		Where("COALESCE(source_batch_no, '') = ''").
		WhereIn(columns.BatchAsyncStatus, []int{consts.BatchAsyncStatusPending, consts.BatchAsyncStatusProcessing}).
		OrderAsc(columns.Id).Limit(50).Scan(&parents)
	if err != nil {
		return cursor, err
	}
	if len(parents) == 0 {
		return 0, nil
	}
	minutes := g.Cfg().MustGet(ctx, "workerBatchAsync.allocateRecovery.processingRetryMinutes", 30).Int()
	if minutes < 30 {
		minutes = 30
	}
	delay := time.Duration(minutes) * time.Minute
	queue := strings.TrimSpace(g.Cfg().MustGet(ctx, "workerBatchAsync.queueName", "iotlcm.batch.execute").String())
	if queue == "" {
		queue = "iotlcm.batch.execute"
	}
	for _, parent := range parents {
		if ctx.Err() != nil {
			return cursor, ctx.Err()
		}
		cursor = parent.Id
		if parent.BatchInfo == nil || parent.BatchInfo.Get("jobRole").String() != "allocate_parent" || !due(parent, time.Now(), delay) {
			continue
		}
		var children []*entity.BatchMain
		err = dao.BatchMain.Ctx(ctx).Where(columns.SourceBatchNo, parent.BatchNo).
			Where(columns.BatchType, consts.BatchTypeAllocate).Where(columns.TenantId, parent.TenantId).
			Where(columns.CreatedBy, parent.CreatedBy).Where(columns.SourceChannel, batchmetadata.MiniProgram).Scan(&children)
		if err != nil {
			return cursor, err
		}
		for _, child := range children {
			if child.BatchInfo == nil || child.BatchInfo.Get("jobRole").String() != "allocate_chunk" || !due(child, time.Now(), delay) {
				continue
			}
			if err := send(ctx, queue, child.BatchNo); err != nil {
				g.Log().Errorf(ctx, "划拨子任务补投失败 batchNo:%s err:%+v", child.BatchNo, err)
			}
		}
		if err := send(ctx, queue, parent.BatchNo); err != nil {
			g.Log().Errorf(ctx, "划拨父任务补投失败 batchNo:%s err:%+v", parent.BatchNo, err)
		}
	}
	if len(parents) < 50 {
		return 0, nil
	}
	return cursor, nil
}

func publish(ctx context.Context, queue, batchNo string) error {
	ctx, cancel := context.WithTimeout(ctx, 10*time.Second)
	defer cancel()
	err := rabbitmq.Publish(ctx, queue, message{BatchNo: batchNo, BatchType: consts.BatchTypeAllocate})
	if err == nil {
		g.Log().Infof(ctx, "划拨持久任务补投 batchNo:%s time:%s", batchNo, gtime.Now())
	}
	return err
}