FranklinWH-Toolbox/service/jobs/emsLogCallSlot.go

package jobs

import (
	"context"
	"errors"
	"fmt"
	"strings"
	"time"

	"fwh-toolbox-server/cache"

	"go.uber.org/zap"
)

const (
	emsLogCallSlotTTL   = 2 * time.Minute
	emsLogCallSlotRetry = time.Second
	EmsLogCallTimeout   = 30 * time.Minute
)

var ErrEmsLogCallTimeout = errors.New("EMS 日志单文件召唤超过 30 分钟")
var ErrEmsLogCallSlotLost = errors.New("EMS 日志召唤名额已丢失")

// EmsLogCallSlot serializes EMS calls for one SN across both download flows and instances.
type EmsLogCallSlot struct {
	lock   *cache.RedisLock
	ctx    context.Context
	cancel context.CancelFunc
	done   chan struct{}
}

func TryAcquireEmsLogCallSlot(ctx context.Context, sn string) (*EmsLogCallSlot, bool, error) {
	keySN := strings.ToUpper(strings.TrimSpace(sn))
	if keySN == "" {
		return nil, false, fmt.Errorf("EMS log SN is empty")
	}
	lock, ok, err := cache.TryLock(ctx, "ems-log:call:"+keySN, emsLogCallSlotTTL)
	if err != nil || !ok {
		return nil, ok, err
	}
	leaseCtx, cancel := context.WithCancel(ctx)
	slot := &EmsLogCallSlot{lock: lock, ctx: leaseCtx, cancel: cancel, done: make(chan struct{})}
	go slot.renew()
	return slot, true, nil
}

func AcquireEmsLogCallSlot(ctx context.Context, sn string) (*EmsLogCallSlot, error) {
	for {
		slot, ok, err := TryAcquireEmsLogCallSlot(ctx, sn)
		if err != nil || ok {
			return slot, err
		}
		select {
		case <-ctx.Done():
			return nil, ctx.Err()
		case <-time.After(emsLogCallSlotRetry):
		}
	}
}

func (s *EmsLogCallSlot) Context() context.Context { return s.ctx }

func (s *EmsLogCallSlot) Release() {
	s.cancel()
	<-s.done
	s.lock.SafeRelease(context.Background())
}

func (s *EmsLogCallSlot) renew() {
	defer close(s.done)
	ticker := time.NewTicker(emsLogCallSlotTTL / 3)
	defer ticker.Stop()
	for {
		select {
		case <-s.ctx.Done():
			return
		case <-ticker.C:
			ctx, cancel := context.WithTimeout(context.Background(), 10*time.Second)
			err := s.lock.Extend(ctx, emsLogCallSlotTTL)
			cancel()
			if err != nil {
				zap.L().Error("EMS 日志召唤名额续期失败", zap.Error(err))
				s.cancel()
				return
			}
		}
	}
}