FranklinWH-Toolbox/service/task/queue/faultRecordQueue/deps.go

package faultRecordQueue

import (
	"context"
	"fmt"
	"time"

	"fwh-toolbox-server/cache/redisx"
	"fwh-toolbox-server/domain"
	"fwh-toolbox-server/repositories/api/fwh_pro"
	"fwh-toolbox-server/repositories/api/fwh_test"
	minioRepo "fwh-toolbox-server/repositories/minio"
	mongoRepo "fwh-toolbox-server/repositories/mongo"
	"fwh-toolbox-server/repositories/mqtt/fwh_mqtt"

	"go.mongodb.org/mongo-driver/bson"
	"go.mongodb.org/mongo-driver/bson/primitive"
	"go.uber.org/zap"
)

// 执行器依赖的外部能力，抽象为接口便于测试

type mqttBus interface {
	Publish(ctx context.Context, topic string, msg fwh_mqtt.FwhMqttMsg) error
	Subscribe(ctx context.Context, topic string, h fwh_mqtt.Handler) (func(), error)
}

type opStore interface {
	Get(ctx context.Context, id primitive.ObjectID) (*domain.FaultRecordOperation, error)
	UpdateStatus(ctx context.Context, id primitive.ObjectID, status domain.FaultRecordOpStatus, extra bson.M) (bool, error)
	SetFields(ctx context.Context, id primitive.ObjectID, fields bson.M) error
	AppendMessage(ctx context.Context, id primitive.ObjectID, msg domain.FaultRecordOpMessage) error
}

type fileStore interface {
	Put(ctx context.Context, key string, data []byte) error
	Get(ctx context.Context, key string) ([]byte, error)
}

type locker interface {
	// Acquire 以 token 加锁并自动续期；返回的 release 停止续期并释放
	Acquire(ctx context.Context, key, token string, ttl time.Duration) (release func(), ok bool, err error)
}

type deps struct {
	bus         func(env string) (mqttBus, error)
	store       opStore
	files       fileStore
	download    func(env, equipNo, fileName string) ([]byte, error)
	downloadURL func(env, equipNo, fileName string) string
	lock        locker
	timeouts    timeouts
}

type timeouts struct {
	ack                 time.Duration // 发送 → 378
	reply               time.Duration // 378 → 379（cmd=0/1/3）
	replyRecord         time.Duration // 378 → 379（cmd=2，设备需读取子设备记录）
	replyPerRecord      time.Duration // cmd=2 每多一条记录增加的等待
	upload              time.Duration // 379 → 609
	uploadPerRecord     time.Duration // 每多一条记录增加的上传等待
	cancelPoll          time.Duration
	lockTTL             time.Duration
	drain               time.Duration // 操作结束后继续记录报文的时间
	afterUploadAck      time.Duration // 收到 610 后等待多久开始下载
	downloadInterval    time.Duration // 下载失败后的重试间隔，一直重试至总超时
	manualRetryAttempts int           // 手动重试下载的最多尝试次数
}

var defaultTimeouts = timeouts{
	ack:                 30 * time.Second,
	reply:               2 * time.Minute,
	replyRecord:         4 * time.Minute,
	replyPerRecord:      time.Minute, // 样本单条约 43 秒
	upload:              5 * time.Minute,
	uploadPerRecord:     30 * time.Second,
	cancelPoll:          time.Second,
	lockTTL:             2 * time.Minute,
	drain:               2 * time.Second,
	afterUploadAck:      15 * time.Second, // 用户要求：收到 610 后 15 秒开始下载
	downloadInterval:    15 * time.Second, // 用户要求：失败后每 15 秒重试
	manualRetryAttempts: 3,                // 用户要求：手动重试只尝试 3 次
}

/************** 默认实现 **************/

type mongoStore struct{}

func (mongoStore) Get(ctx context.Context, id primitive.ObjectID) (*domain.FaultRecordOperation, error) {
	return mongoRepo.GetFaultRecordOp(ctx, id)
}
func (mongoStore) UpdateStatus(ctx context.Context, id primitive.ObjectID, s domain.FaultRecordOpStatus, extra bson.M) (bool, error) {
	return mongoRepo.UpdateFaultRecordOpStatus(ctx, id, s, extra)
}
func (mongoStore) SetFields(ctx context.Context, id primitive.ObjectID, f bson.M) error {
	return mongoRepo.SetFaultRecordOpFields(ctx, id, f)
}
func (mongoStore) AppendMessage(ctx context.Context, id primitive.ObjectID, m domain.FaultRecordOpMessage) error {
	return mongoRepo.AppendFaultRecordOpMessage(ctx, id, m)
}

type minioStore struct{}

func (minioStore) Put(ctx context.Context, key string, data []byte) error {
	return minioRepo.PutFaultRecordFile(ctx, key, data)
}
func (minioStore) Get(ctx context.Context, key string) ([]byte, error) {
	return minioRepo.ReadFaultRecordFile(ctx, key)
}

func mqttBusFor(env string) (mqttBus, error) {
	e, err := fwh_mqtt.ParseEnv(env)
	if err != nil {
		return nil, err
	}
	return fwh_mqtt.GetClient(e)
}

func downloadURLFor(env, equipNo, fileName string) string {
	if env == string(fwh_mqtt.EnvTest) {
		return fwh_test.FaultRecordUploadFileURL(equipNo, fileName)
	}
	return fwh_pro.FaultRecordUploadFileURL(equipNo, fileName)
}

func downloadFor(env, equipNo, fileName string) ([]byte, error) {
	switch env {
	case string(fwh_mqtt.EnvProd):
		return fwh_pro.DownloadFaultRecordUploadFile(equipNo, fileName)
	case string(fwh_mqtt.EnvTest):
		return fwh_test.DownloadFaultRecordUploadFile(equipNo, fileName)
	}
	return nil, fmt.Errorf("unknown env %s", env)
}

type redisLocker struct{}

func (redisLocker) Acquire(ctx context.Context, key, token string, ttl time.Duration) (func(), bool, error) {
	lock, ok, err := redisx.TryLockWithToken(ctx, key, token, ttl)
	if err != nil || !ok {
		return nil, ok, err
	}
	stop := make(chan struct{})
	go func() {
		t := time.NewTicker(ttl / 3)
		defer t.Stop()
		for {
			select {
			case <-stop:
				return
			case <-t.C:
				if err := lock.Extend(context.Background(), ttl); err != nil {
					zap.L().Warn("fault record gateway lock extend failed", zap.String("key", key), zap.Error(err))
				}
			}
		}
	}()
	return func() {
		close(stop)
		lock.SafeRelease(context.Background())
	}, true, nil
}

func defaultDeps() deps {
	return deps{
		bus:         mqttBusFor,
		store:       mongoStore{},
		files:       minioStore{},
		download:    downloadFor,
		downloadURL: downloadURLFor,
		lock:        redisLocker{},
		timeouts:    defaultTimeouts,
	}
}