sync: 对齐小程序 v1.0.2 生产代码并脱敏

2026-10-29 15:15:12 +08:00
parent a7abd476f0
commit 5fcf74deb0
423 changed files with 15001 additions and 6382 deletions
@@ -0,0 +1,144 @@
+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
+}