feat: rbac完善，file接入完成

2026-05-01 12:56:08 +08:00
parent bbd3f834b9
commit a93477ea8e
81 changed files with 5470 additions and 371 deletions
@@ -0,0 +1,58 @@
+package oss_core
+
+import (
+	"context"
+	"fmt"
+	"io"
+	"mime/multipart"
+	"path/filepath"
+	"strings"
+	"time"
+
+	"github.com/aliyun/aliyun-oss-go-sdk/oss"
+	"sundynix-micro-go/app/file/rpc/file"
+	"sundynix-micro-go/common/utils/hash"
+)
+
+type AliyunUploader struct {
+	bucket *oss.Bucket
+	config *file.StorageConfigInfo
+}
+
+func NewAliyunUploader(conf *file.StorageConfigInfo) (Uploader, error) {
+	client, err := oss.New(conf.Endpoint, conf.AccessKeyId, conf.AccessKeySecret)
+	if err != nil {
+		return nil, err
+	}
+	bucket, err := client.Bucket(conf.BucketName)
+	if err != nil {
+		return nil, err
+	}
+	return &AliyunUploader{
+		bucket: bucket,
+		config: conf,
+	}, nil
+}
+
+func (a *AliyunUploader) UploadFile(ctx context.Context, file multipart.File, fileHeader *multipart.FileHeader) (string, string, error) {
+	ext := filepath.Ext(fileHeader.Filename)
+	filename := hash.MD5([]byte(strings.TrimSuffix(fileHeader.Filename, ext))) + ext
+	timestr := fmt.Sprintf("%d", time.Now().UnixMicro())
+	objectKey := time.Now().Format("2006-01-02") + "/" + timestr + "-" + filename
+
+	err := a.bucket.PutObject(objectKey, file)
+	if err != nil {
+		return "", "", fmt.Errorf("上传阿里云OSS失败: %v", err)
+	}
+
+	url := a.config.BucketUrl + "/" + objectKey
+	return url, objectKey, nil
+}
+
+func (a *AliyunUploader) DeleteFile(ctx context.Context, key string) error {
+	return a.bucket.DeleteObject(key)
+}
+
+func (a *AliyunUploader) DownloadFile(ctx context.Context, key string) (io.ReadCloser, error) {
+	return a.bucket.GetObject(key)
+}
@@ -0,0 +1,89 @@
+package oss_core
+
+import (
+	"context"
+	"fmt"
+	"reflect"
+	"sundynix-micro-go/app/file/rpc/file"
+	"sundynix-micro-go/app/file/rpc/fileservice"
+	"sync"
+)
+
+type CachedUploader struct {
+	Uploader Uploader
+	Config   *file.StorageConfigInfo
+}
+
+var (
+	uploaderCache sync.Map // map[string]*CachedUploader
+	cacheMutex    sync.Mutex
+)
+
+// OSSFactory 工厂结构体
+type OSSFactory struct {
+	fileRpc fileservice.FileService
+}
+
+func NewOSSFactory(fileRpc fileservice.FileService) *OSSFactory {
+	return &OSSFactory{
+		fileRpc: fileRpc,
+	}
+}
+
+// GetActiveUploader 获取当前激活的存储上传实例
+func (f *OSSFactory) GetActiveUploader(ctx context.Context) (Uploader, error) {
+	resp, err := f.fileRpc.GetDefaultStorageConfig(ctx, &fileservice.GetDefaultStorageConfigReq{})
+	if err != nil || resp.Config == nil {
+		return nil, fmt.Errorf("未找到激活的存储配置: %v", err)
+	}
+
+	conf := resp.Config
+
+	// 1. 尝试从缓存获取
+	if cachedVal, ok := uploaderCache.Load(conf.Id); ok {
+		cachedItem := cachedVal.(*CachedUploader)
+		// 优雅之处：深度对比配置是否发生变化，完美支持前端动态修改配置后实时热加载
+		if reflect.DeepEqual(cachedItem.Config, conf) {
+			return cachedItem.Uploader, nil
+		}
+	}
+
+	// 2. 加锁防止高并发下的重复实例化
+	cacheMutex.Lock()
+	defer cacheMutex.Unlock()
+
+	// 3. Double-Check
+	if cachedVal, ok := uploaderCache.Load(conf.Id); ok {
+		cachedItem := cachedVal.(*CachedUploader)
+		if reflect.DeepEqual(cachedItem.Config, conf) {
+			return cachedItem.Uploader, nil
+		}
+	}
+
+	// 4. 实例化新的 Uploader
+	var uploader Uploader
+	switch conf.Type {
+	case "minio":
+		uploader, err = NewMinioUploader(conf)
+	case "aliyun":
+		uploader, err = NewAliyunUploader(conf)
+	case "tencent":
+		uploader, err = NewTencentUploader(conf)
+	case "qiniu":
+		uploader, err = NewQiniuUploader(conf)
+	default:
+		return nil, fmt.Errorf("不支持的存储类型: %s", conf.Type)
+	}
+
+	if err != nil {
+		return nil, err
+	}
+
+	// 5. 更新缓存
+	uploaderCache.Store(conf.Id, &CachedUploader{
+		Uploader: uploader,
+		Config:   conf,
+	})
+
+	return uploader, nil
+}
@@ -0,0 +1,95 @@
+package oss_core
+
+import (
+	"context"
+	"fmt"
+	"io"
+	"mime/multipart"
+	"path/filepath"
+	"strings"
+	"time"
+
+	"github.com/minio/minio-go/v7"
+	"github.com/minio/minio-go/v7/pkg/credentials"
+	"sundynix-micro-go/app/file/rpc/file"
+	"sundynix-micro-go/common/utils/hash"
+)
+
+type MinioUploader struct {
+	client *minio.Client
+	config *file.StorageConfigInfo
+}
+
+func NewMinioUploader(conf *file.StorageConfigInfo) (Uploader, error) {
+	// 判断如果是 https 或者是类似云服务，可以根据 endpoint 后缀猜测 secure
+	useSSL := strings.HasPrefix(conf.Endpoint, "https://") || strings.Contains(conf.Endpoint, "aliyuncs.com") || conf.Endpoint == "oss-cn-hangzhou.aliyuncs.com" // 这里只是示例
+	if strings.HasPrefix(conf.Endpoint, "http://") || strings.HasPrefix(conf.Endpoint, "https://") {
+		// 移除协议头给 minio 客户端
+		conf.Endpoint = strings.TrimPrefix(conf.Endpoint, "http://")
+		conf.Endpoint = strings.TrimPrefix(conf.Endpoint, "https://")
+	} else {
+		// 默认非 SSL
+		useSSL = false
+	}
+
+	client, err := minio.New(conf.Endpoint, &minio.Options{
+		Creds:  credentials.NewStaticV4(conf.AccessKeyId, conf.AccessKeySecret, ""),
+		Secure: useSSL,
+	})
+	if err != nil {
+		return nil, err
+	}
+
+	// 这里可以加上检查 bucket
+	return &MinioUploader{
+		client: client,
+		config: conf,
+	}, nil
+}
+
+func (m *MinioUploader) UploadFile(ctx context.Context, file multipart.File, fileHeader *multipart.FileHeader) (string, string, error) {
+	// 直接重置文件指针（保险起见）
+	file.Seek(0, 0)
+
+	ext := filepath.Ext(fileHeader.Filename)
+	filename := hash.MD5([]byte(strings.TrimSuffix(fileHeader.Filename, ext))) + ext
+	timestr := fmt.Sprintf("%d", time.Now().UnixMicro())
+	objectKey := time.Now().Format("2006-01-02") + "/" + timestr + "-" + filename
+
+	bucketName := m.config.BucketName
+
+	// 直接执行 PutObject（流式上传，不再全部读入内存）
+	info, err := m.client.PutObject(ctx, bucketName, objectKey, file, fileHeader.Size, minio.PutObjectOptions{
+		ContentType: "application/octet-stream",
+	})
+
+	if err != nil {
+		// 如果是因为 bucket 不存在，则尝试创建并重试
+		if minio.ToErrorResponse(err).Code == "NoSuchBucket" {
+			m.client.MakeBucket(ctx, bucketName, minio.MakeBucketOptions{})
+			// 重置文件流位置，准备重试
+			file.Seek(0, 0)
+			info, err = m.client.PutObject(ctx, bucketName, objectKey, file, fileHeader.Size, minio.PutObjectOptions{
+				ContentType: "application/octet-stream",
+			})
+		}
+		if err != nil {
+			return "", "", fmt.Errorf("上传Minio失败: %v", err)
+		}
+	}
+
+	url := m.config.BucketUrl + "/" + info.Key
+	return url, info.Key, nil
+}
+
+func (m *MinioUploader) DeleteFile(ctx context.Context, key string) error {
+	return m.client.RemoveObject(ctx, m.config.BucketName, key, minio.RemoveObjectOptions{})
+}
+
+func (m *MinioUploader) DownloadFile(ctx context.Context, key string) (io.ReadCloser, error) {
+	obj, err := m.client.GetObject(ctx, m.config.BucketName, key, minio.GetObjectOptions{})
+	if err != nil {
+		return nil, err
+	}
+	return obj, nil
+}
@@ -0,0 +1,84 @@
+package oss_core
+
+import (
+	"context"
+	"fmt"
+	"io"
+	"mime/multipart"
+	"net/http"
+	"path/filepath"
+	"strings"
+	"time"
+
+	"github.com/qiniu/go-sdk/v7/auth/qbox"
+	"github.com/qiniu/go-sdk/v7/storage"
+	"sundynix-micro-go/app/file/rpc/file"
+	"sundynix-micro-go/common/utils/hash"
+)
+
+type QiniuUploader struct {
+	mac    *qbox.Mac
+	config *file.StorageConfigInfo
+}
+
+func NewQiniuUploader(conf *file.StorageConfigInfo) (Uploader, error) {
+	mac := qbox.NewMac(conf.AccessKeyId, conf.AccessKeySecret)
+	return &QiniuUploader{
+		mac:    mac,
+		config: conf,
+	}, nil
+}
+
+func (q *QiniuUploader) UploadFile(ctx context.Context, file multipart.File, fileHeader *multipart.FileHeader) (string, string, error) {
+	ext := filepath.Ext(fileHeader.Filename)
+	filename := hash.MD5([]byte(strings.TrimSuffix(fileHeader.Filename, ext))) + ext
+	timestr := fmt.Sprintf("%d", time.Now().UnixMicro())
+	objectKey := time.Now().Format("2006-01-02") + "/" + timestr + "-" + filename
+
+	putPolicy := storage.PutPolicy{
+		Scope: q.config.BucketName,
+	}
+	upToken := putPolicy.UploadToken(q.mac)
+
+	cfg := storage.Config{}
+	// 根据配置中的 Region 判断
+	// 这里简单写死，如果要自适应可以根据 config.Region 给定
+	cfg.Zone = &storage.ZoneHuadong
+	cfg.UseHTTPS = false
+	cfg.UseCdnDomains = false
+
+	formUploader := storage.NewFormUploader(&cfg)
+	ret := storage.PutRet{}
+
+	err := formUploader.Put(ctx, &ret, upToken, objectKey, file, fileHeader.Size, nil)
+	if err != nil {
+		return "", "", fmt.Errorf("上传七牛云失败: %v", err)
+	}
+
+	url := q.config.BucketUrl + "/" + ret.Key
+	return url, ret.Key, nil
+}
+
+func (q *QiniuUploader) DeleteFile(ctx context.Context, key string) error {
+	cfg := storage.Config{}
+	cfg.Zone = &storage.ZoneHuadong
+	bucketManager := storage.NewBucketManager(q.mac, &cfg)
+	return bucketManager.Delete(q.config.BucketName, key)
+}
+
+func (q *QiniuUploader) DownloadFile(ctx context.Context, key string) (io.ReadCloser, error) {
+	mac := qbox.NewMac(q.config.AccessKeyId, q.config.AccessKeySecret)
+	domain := q.config.BucketUrl
+	deadline := time.Now().Add(time.Second * 3600).Unix() // 1小时有效期
+	privateAccessURL := storage.MakePrivateURL(mac, domain, key, deadline)
+
+	resp, err := http.Get(privateAccessURL)
+	if err != nil {
+		return nil, err
+	}
+	if resp.StatusCode != http.StatusOK {
+		resp.Body.Close()
+		return nil, fmt.Errorf("qiniu download failed with status: %s", resp.Status)
+	}
+	return resp.Body, nil
+}
@@ -0,0 +1,73 @@
+package oss_core
+
+import (
+	"context"
+	"fmt"
+	"io"
+	"mime/multipart"
+	"net/http"
+	"net/url"
+	"path/filepath"
+	"strings"
+	"time"
+
+	"github.com/tencentyun/cos-go-sdk-v5"
+	"sundynix-micro-go/app/file/rpc/file"
+	"sundynix-micro-go/common/utils/hash"
+)
+
+type TencentUploader struct {
+	client *cos.Client
+	config *file.StorageConfigInfo
+}
+
+func NewTencentUploader(conf *file.StorageConfigInfo) (Uploader, error) {
+	// Endpoint should be something like https://bucket-appid.cos.ap-guangzhou.myqcloud.com
+	// But usually users just put bucket name and region, here we assume endpoint is full bucket url
+	// Or we use bucketUrl as the endpoint for client initialization if endpoint is not formatted well
+	u, err := url.Parse(conf.Endpoint)
+	if err != nil {
+		return nil, fmt.Errorf("解析Endpoint失败: %v", err)
+	}
+
+	b := &cos.BaseURL{BucketURL: u}
+	client := cos.NewClient(b, &http.Client{
+		Transport: &cos.AuthorizationTransport{
+			SecretID:  conf.AccessKeyId,
+			SecretKey: conf.AccessKeySecret,
+		},
+	})
+
+	return &TencentUploader{
+		client: client,
+		config: conf,
+	}, nil
+}
+
+func (t *TencentUploader) UploadFile(ctx context.Context, file multipart.File, fileHeader *multipart.FileHeader) (string, string, error) {
+	ext := filepath.Ext(fileHeader.Filename)
+	filename := hash.MD5([]byte(strings.TrimSuffix(fileHeader.Filename, ext))) + ext
+	timestr := fmt.Sprintf("%d", time.Now().UnixMicro())
+	objectKey := time.Now().Format("2006-01-02") + "/" + timestr + "-" + filename
+
+	_, err := t.client.Object.Put(ctx, objectKey, file, nil)
+	if err != nil {
+		return "", "", fmt.Errorf("上传腾讯云COS失败: %v", err)
+	}
+
+	fileUrl := t.config.BucketUrl + "/" + objectKey
+	return fileUrl, objectKey, nil
+}
+
+func (t *TencentUploader) DeleteFile(ctx context.Context, key string) error {
+	_, err := t.client.Object.Delete(ctx, key)
+	return err
+}
+
+func (t *TencentUploader) DownloadFile(ctx context.Context, key string) (io.ReadCloser, error) {
+	resp, err := t.client.Object.Get(ctx, key, nil)
+	if err != nil {
+		return nil, err
+	}
+	return resp.Body, nil
+}
@@ -0,0 +1,17 @@
+package oss_core
+
+import (
+	"context"
+	"io"
+	"mime/multipart"
+)
+
+// Uploader 统一存储上传接口
+type Uploader interface {
+	// UploadFile 接收文件流并上传，返回存储的具体url，标识key和可能的错误
+	UploadFile(ctx context.Context, file multipart.File, fileHeader *multipart.FileHeader) (url string, key string, err error)
+	// DeleteFile 删除远端存储的文件
+	DeleteFile(ctx context.Context, key string) error
+	// DownloadFile 获取文件下载的数据流
+	DownloadFile(ctx context.Context, key string) (io.ReadCloser, error)
+}