hadja_server/web/service/srv_mediaresource/cover_backfill.go

package srv_mediaresource

import (
	"context"
	"encoding/json"
	"fmt"
	"strings"
	"time"
	"videoweb/pub/common/log"
	"videoweb/pub/models/playletmod"
	"videoweb/pub/pubdef"
	"videoweb/pub/redis"

	goredis "github.com/go-redis/redis"
	"go.mongodb.org/mongo-driver/bson"
	"go.mongodb.org/mongo-driver/mongo/options"
)

const (
	// PlayletCoverCompressCronSpec 历史空压缩封面扫描频率：每5分钟投递一批
	PlayletCoverCompressCronSpec = "0 */5 * * * ?"

	playletCoverCompressQueueKeyFmt = "playlet:cover:compress:queue:%d"
	playletCoverCompressLockKeyFmt  = "playlet:cover:compress:lock:%d:%d"
	playletCoverCompressFailKeyFmt  = "playlet:cover:compress:fail:v2:%d:%d"
	playletCoverCompressLockTTL     = 10 * time.Minute
	playletCoverCompressFailTTL     = 10 * time.Minute
	playletCoverCompressPopWait     = 5 * time.Second
	playletCoverCompressRetryWait   = time.Second
	playletCoverCompressScanSize    = int64(200)
	playletCoverCompressBatchSize   = 50
)

func playletCoverCompressQueueKey() string {
	return fmt.Sprintf(playletCoverCompressQueueKeyFmt, pubdef.AppId)
}

func playletCoverCompressLockKey(playletId int64) string {
	return fmt.Sprintf(playletCoverCompressLockKeyFmt, pubdef.AppId, playletId)
}

func playletCoverCompressFailKey(playletId int64) string {
	return fmt.Sprintf(playletCoverCompressFailKeyFmt, pubdef.AppId, playletId)
}

type playletCoverCompressJob struct {
	PlayletId  int64 `json:"playletId"`
	CreateTime int64 `json:"createTime"`
}

// EnqueuePlayletCoverCompressJobs 扫描压缩封面为空的短剧并投递后台任务
func EnqueuePlayletCoverCompressJobs() {
	ctx := context.Background()
	enqueued := 0
	scanned := 0
	var lastId int64
	for enqueued < playletCoverCompressBatchSize {
		filter := emptyCompressedCoverFilter()
		if lastId > 0 {
			filter[playletmod.FN.Playlet.Id] = bson.M{"$gt": lastId}
		}
		opts := options.Find().
			SetProjection(bson.D{
				{Key: playletmod.FN.Playlet.Id, Value: 1},
				{Key: playletmod.FN.Playlet.Cover, Value: 1},
				{Key: playletmod.FN.Playlet.CompressedCover, Value: 1},
				{Key: playletmod.FN.Playlet.Status, Value: 1},
			}).
			SetSort(bson.D{{Key: playletmod.FN.Playlet.Id, Value: 1}}).
			SetLimit(playletCoverCompressScanSize)
		list, err := playletmod.FindManyPlaylet(ctx, filter, opts)
		if err != nil {
			log.Error("扫描待压缩短剧封面失败", log.E(err))
			return
		}
		if len(list) == 0 {
			break
		}
		scanned += len(list)
		for _, item := range list {
			if item == nil || item.Id <= 0 {
				continue
			}
			lastId = item.Id
			if enqueued >= playletCoverCompressBatchSize {
				break
			}
			if has, _ := redis.Exists(playletCoverCompressFailKey(item.Id)); has {
				continue
			}
			if err := pushPlayletCoverCompressJob(ctx, item.Id); err != nil {
				log.Error("投递短剧封面压缩任务失败", log.Any("playletId", item.Id), log.E(err))
				continue
			}
			enqueued++
		}
		if int64(len(list)) < playletCoverCompressScanSize {
			break
		}
	}
	log.Info("短剧压缩封面任务扫描完成", log.Any("scanCount", scanned), log.Any("enqueued", enqueued))
}

func emptyCompressedCoverFilter() bson.M {
	return bson.M{
		playletmod.FN.Playlet.Cover: bson.M{"$nin": []interface{}{"", nil}},
		"$or": []bson.M{
			{playletmod.FN.Playlet.CompressedCover: bson.M{"$exists": false}},
			{playletmod.FN.Playlet.CompressedCover: ""},
			{playletmod.FN.Playlet.CompressedCover: nil},
		},
	}
}

func pushPlayletCoverCompressJob(ctx context.Context, playletId int64) error {
	select {
	case <-ctx.Done():
		return ctx.Err()
	default:
	}
	payload, err := json.Marshal(playletCoverCompressJob{
		PlayletId:  playletId,
		CreateTime: time.Now().Unix(),
	})
	if err != nil {
		return err
	}
	return redis.RPush(playletCoverCompressQueueKey(), string(payload))
}

func runPlayletCoverCompressWorker(ctx context.Context) {
	for {
		select {
		case <-ctx.Done():
			return
		default:
		}
		result, err := redis.Client().BLPop(playletCoverCompressPopWait, playletCoverCompressQueueKey()).Result()
		if err != nil {
			if err == goredis.Nil {
				continue
			}
			log.Error("消费短剧封面压缩任务失败", log.E(err))
			time.Sleep(playletCoverCompressRetryWait)
			continue
		}
		if len(result) < 2 {
			log.Error("短剧封面压缩任务格式异常", log.Any("result", result))
			continue
		}
		handlePlayletCoverCompressPayload(ctx, result[1])
	}
}

func handlePlayletCoverCompressPayload(ctx context.Context, payload string) {
	defer func() {
		if err := recover(); err != nil {
			log.Error("处理短剧封面压缩任务异常", log.Any("err", err), log.Any("payload", payload))
		}
	}()
	var job playletCoverCompressJob
	if err := json.Unmarshal([]byte(payload), &job); err != nil {
		log.Error("解析短剧封面压缩任务失败", log.Any("payload", payload), log.E(err))
		return
	}
	if job.PlayletId <= 0 {
		return
	}
	locked, err := redis.Setnx_NewOK(playletCoverCompressLockKey(job.PlayletId), true, playletCoverCompressLockTTL)
	if err != nil {
		log.Error("获取短剧封面压缩锁失败", log.Any("playletId", job.PlayletId), log.E(err))
		return
	}
	if !locked {
		return
	}
	defer func() {
		if err := redis.Del(playletCoverCompressLockKey(job.PlayletId)); err != nil {
			log.Error("释放短剧封面压缩锁失败", log.Any("playletId", job.PlayletId), log.E(err))
		}
	}()
	playlet, serr := playletmod.FindOnePlaylet(ctx, bson.M{playletmod.FN.Playlet.Id: job.PlayletId}, options.FindOne().SetProjection(bson.D{
		{Key: playletmod.FN.Playlet.Id, Value: 1},
		{Key: playletmod.FN.Playlet.Cover, Value: 1},
		{Key: playletmod.FN.Playlet.CompressedCover, Value: 1},
		{Key: playletmod.FN.Playlet.Status, Value: 1},
	}))
	if serr != nil || playlet == nil {
		if serr != nil {
			log.Warn("查询待压缩短剧失败", log.Any("playletId", job.PlayletId), log.E(serr))
		}
		return
	}
	if FillPlayletCompressedCover(ctx, playlet) {
		log.Info("历史短剧压缩封面完成", log.Any("playletId", playlet.Id))
		return
	}
	if strings.TrimSpace(playlet.CompressedCover) == "" && strings.TrimSpace(playlet.Cover) != "" {
		if err := redis.Set(playletCoverCompressFailKey(playlet.Id), "1", playletCoverCompressFailTTL); err != nil {
			log.Warn("记录短剧封面压缩失败冷却失败", log.Any("playletId", playlet.Id), log.E(err))
		}
	}
}