VptPassiveAdapter/fs/ali_adapter.go

package fs

import (
	"ZhenTuLocalPassiveAdapter/config"
	"ZhenTuLocalPassiveAdapter/dto"
	"ZhenTuLocalPassiveAdapter/logger"
	"ZhenTuLocalPassiveAdapter/util"
	"context"
	"fmt"
	"path"
	"sort"
	"sync"
	"time"

	"github.com/aliyun/aliyun-oss-go-sdk/oss"
	"go.opentelemetry.io/otel/attribute"
	"go.opentelemetry.io/otel/codes"
	"go.uber.org/zap"
)

var aliOssCache sync.Map

type AliOSSAdapter struct {
	StorageConfig config.StorageConfig
	ossClient     *oss.Client
}

func (a *AliOSSAdapter) getClient() (*oss.Client, error) {
	if a.ossClient == nil {
		client, err := oss.New(
			a.StorageConfig.AliOSS.Endpoint,
			a.StorageConfig.AliOSS.AccessKeyId,
			a.StorageConfig.AliOSS.AccessKeySecret,
		)
		if err != nil {
			return nil, fmt.Errorf("创建阿里云OSS客户端失败: %w", err)
		}
		a.ossClient = client
	}
	return a.ossClient, nil
}

func (a *AliOSSAdapter) GetFileList(ctx context.Context, dirPath string, relDt time.Time) ([]dto.File, error) {
	_, span := tracer.Start(ctx, "GetFileList_alioss")
	defer span.End()

	span.SetAttributes(attribute.String("path", dirPath))
	span.SetAttributes(attribute.String("relativeDate", relDt.Format("2006-01-02")))
	if a.StorageConfig.AliOSS.Bucket == "" {
		span.SetAttributes(attribute.String("error", "未配置阿里云OSS存储桶"))
		span.SetStatus(codes.Error, "未配置阿里云OSS存储桶")
		return nil, fmt.Errorf("未配置阿里云OSS存储桶")
	}

	cacheKey := fmt.Sprintf("%s_%s", dirPath, relDt.Format("2006-01-02"))
	if cachedInterface, ok := aliOssCache.Load(cacheKey); ok {
		cachedItem := cachedInterface.(cacheItem)
		logger.Debug("缓存过期时间", zap.Duration("expiresIn", cachedItem.expires.Sub(time.Now())))
		if time.Now().Before(cachedItem.expires) {
			logger.Debug("获取已缓存列表", zap.String("cacheKey", cacheKey))
			span.SetAttributes(attribute.Bool("cache.hit", true))
			return cachedItem.data, nil
		}
	}

	mutexKey := fmt.Sprintf("lock_%s", cacheKey)
	mutex, _ := aliOssCache.LoadOrStore(mutexKey, &sync.Mutex{})
	lock := mutex.(*sync.Mutex)
	defer func() {
		// 解锁后删除锁(避免内存泄漏)
		aliOssCache.Delete(mutexKey)
		lock.Unlock()
	}()
	lock.Lock()

	if cachedInterface, ok := aliOssCache.Load(cacheKey); ok {
		cachedItem := cachedInterface.(cacheItem)
		logger.Debug("缓存过期时间", zap.Duration("expiresIn", cachedItem.expires.Sub(time.Now())))
		if time.Now().Before(cachedItem.expires) {
			logger.Debug("过锁后获取已缓存列表", zap.String("cacheKey", cacheKey))
			span.SetAttributes(attribute.Bool("aliOssCache.hit", true))
			return cachedItem.data, nil
		}
	}

	client, err := a.getClient()
	if err != nil {
		span.SetAttributes(attribute.String("error", err.Error()))
		span.SetStatus(codes.Error, "创建阿里云OSS客户端失败")
		return nil, err
	}

	bucket, err := client.Bucket(a.StorageConfig.AliOSS.Bucket)
	if err != nil {
		span.SetAttributes(attribute.String("error", err.Error()))
		span.SetStatus(codes.Error, "获取存储桶失败")
		return nil, fmt.Errorf("获取存储桶失败: %w", err)
	}

	var fileList []dto.File
	prefix := path.Join(a.StorageConfig.AliOSS.Prefix, dirPath)
	marker := ""

	for {
		lsRes, err := bucket.ListObjects(
			oss.Prefix(prefix),
			oss.Marker(marker),
			oss.MaxKeys(1000),
		)
		if err != nil {
			span.SetAttributes(attribute.String("error", err.Error()))
			span.SetStatus(codes.Error, "文件列表读取失败")
			return nil, fmt.Errorf("文件列表读取失败: %w", err)
		}

		for _, object := range lsRes.Objects {
			key := object.Key
			if !util.IsVideoFile(path.Base(key)) {
				continue
			}
			startTime, stopTime, err := util.ParseStartStopTime(path.Base(key), relDt)
			if err != nil {
				continue
			}
			if stopTime.IsZero() {
				stopTime = startTime
			}
			if startTime.Equal(stopTime) {
				stopTime = stopTime.Add(time.Second * time.Duration(config.Config.Record.Duration))
			}

			// 生成预签名URL(有效期10分钟)
			signedURL, err := bucket.SignURL(key, oss.HTTPGet, 600)
			if err != nil {
				span.SetAttributes(attribute.String("error", err.Error()))
				span.SetStatus(codes.Error, "生成预签名URL失败")
				logger.Error("生成预签名URL失败", zap.Error(err))
				continue
			}

			fileList = append(fileList, dto.File{
				BasePath:  a.StorageConfig.AliOSS.Bucket,
				Name:      path.Base(key),
				Path:      path.Dir(key),
				Url:       signedURL,
				StartTime: startTime,
				EndTime:   stopTime,
			})
		}

		if !lsRes.IsTruncated {
			break
		}
		marker = lsRes.NextMarker
	}

	span.SetAttributes(attribute.Int("file.count", len(fileList)))
	sort.Slice(fileList, func(i, j int) bool {
		return fileList[i].StartTime.Before(fileList[j].StartTime)
	})
	span.SetStatus(codes.Ok, "文件读取成功")

	cacheItem := cacheItem{
		data:    fileList,
		expires: time.Now().Add(30 * time.Second),
	}
	aliOssCache.Store(cacheKey, cacheItem)
	logger.Debug("缓存文件列表", zap.String("cacheKey", cacheKey))

	return fileList, nil
}

// 添加定时清理缓存的初始化函数
func init() {
	go func() {
		ticker := time.NewTicker(1 * time.Minute)
		defer ticker.Stop()
		for {
			select {
			case <-ticker.C:
				cleanupAliOssCache()
			}
		}
	}()
}

// 添加缓存清理函数
func cleanupAliOssCache() {
	var keysToDelete []interface{}
	aliOssCache.Range(func(key, value interface{}) bool {
		item, ok := value.(cacheItem)
		if !ok {
			return true
		}
		if time.Now().After(item.expires) {
			keysToDelete = append(keysToDelete, key)
		}
		return true
	})
	for _, key := range keysToDelete {
		aliOssCache.Delete(key)
	}
}