You can not select more than 25 topics Topics must start with a letter or number, can include dashes ('-') and can be up to 35 characters long.
Open-IM-Server/pkg/common/db/s3/cont/controller.go

258 lines
6.6 KiB

merge v3dev into main (#504) * statistics user register * refactor: router change * minio init * UserRegisterCount * push use local conn * refactor: user pb update * remove online push close grpc conn * refactor: user pb update * refactor:pb file * msgs statistics * msgs statistics * revoke userID * refactor: errcode update * active user * active user * active user * refactor: errcode update * feat: conn update token * active user * active user * feat: conn update token * active user * feat: conn update token * feat: conn update token * feat: conn update token * add tx_oss cos * active user * active user * group create * group create * feat: group notification show to conversation * feat: group notification show to conversation * group active * user active * sendNotificationWithName * withname * privateChat * a2r call option * grpc with detail return error * change log error * chain unary interceptor * api nil slice map * fix sync has read * fix: text update * fix: update add model * set conversations update * set privateChat * fix: content update * remove unuse rpc * msgDestruct * cron use rpc mw * set IsMsgDestruct * msg destruct * msgDestruct * s3 minio, cos, oss support * feat: add implement of GetUsersOnlineStatus, #472 (#477) * s3 minio, cos, oss support * s3 route * remove extendMsg code * s3 route * remove unuse code * s3 pb * s3 pb * s3 pb * s3 presigned put * s3 presigned test * s3 presigned test * s3 presigned test * s3 presigned test * s3 presigned test * s3 presigned test * s3 presigned test * s3 presigned test * Update .gitignore (#482) * s3 debug log * s3 debug log * cron add log and fix cron * add log * cron * s3 config * fix kick user bug * s3 cos * add kick log * s3 cos test * s3 cos test * s3 cos test * kick user log * kickuserlog * s3 cos copy * s3 cos copy * s3 url * s3 url * s3 AccessURL * log * s3 InitiateMultipartUpload add ExpireTime --------- Co-authored-by: withchao <993506633@qq.com> Co-authored-by: wangchuxiao <wangchuxiao97@outlook.com> Co-authored-by: BanTanger <88583317+BanTanger@users.noreply.github.com> Co-authored-by: withchao <48119764+withchao@users.noreply.github.com> Co-authored-by: Alan <68671759+hanzhixiao@users.noreply.github.com>
1 year ago
package cont
import (
"context"
"crypto/md5"
"encoding/hex"
"errors"
"fmt"
"path"
"strings"
"time"
"github.com/google/uuid"
"github.com/OpenIMSDK/Open-IM-Server/pkg/common/db/s3"
"github.com/OpenIMSDK/Open-IM-Server/pkg/common/log"
"github.com/OpenIMSDK/Open-IM-Server/pkg/errs"
merge v3dev into main (#504) * statistics user register * refactor: router change * minio init * UserRegisterCount * push use local conn * refactor: user pb update * remove online push close grpc conn * refactor: user pb update * refactor:pb file * msgs statistics * msgs statistics * revoke userID * refactor: errcode update * active user * active user * active user * refactor: errcode update * feat: conn update token * active user * active user * feat: conn update token * active user * feat: conn update token * feat: conn update token * feat: conn update token * add tx_oss cos * active user * active user * group create * group create * feat: group notification show to conversation * feat: group notification show to conversation * group active * user active * sendNotificationWithName * withname * privateChat * a2r call option * grpc with detail return error * change log error * chain unary interceptor * api nil slice map * fix sync has read * fix: text update * fix: update add model * set conversations update * set privateChat * fix: content update * remove unuse rpc * msgDestruct * cron use rpc mw * set IsMsgDestruct * msg destruct * msgDestruct * s3 minio, cos, oss support * feat: add implement of GetUsersOnlineStatus, #472 (#477) * s3 minio, cos, oss support * s3 route * remove extendMsg code * s3 route * remove unuse code * s3 pb * s3 pb * s3 pb * s3 presigned put * s3 presigned test * s3 presigned test * s3 presigned test * s3 presigned test * s3 presigned test * s3 presigned test * s3 presigned test * s3 presigned test * Update .gitignore (#482) * s3 debug log * s3 debug log * cron add log and fix cron * add log * cron * s3 config * fix kick user bug * s3 cos * add kick log * s3 cos test * s3 cos test * s3 cos test * kick user log * kickuserlog * s3 cos copy * s3 cos copy * s3 url * s3 url * s3 AccessURL * log * s3 InitiateMultipartUpload add ExpireTime --------- Co-authored-by: withchao <993506633@qq.com> Co-authored-by: wangchuxiao <wangchuxiao97@outlook.com> Co-authored-by: BanTanger <88583317+BanTanger@users.noreply.github.com> Co-authored-by: withchao <48119764+withchao@users.noreply.github.com> Co-authored-by: Alan <68671759+hanzhixiao@users.noreply.github.com>
1 year ago
)
func New(impl s3.Interface) *Controller {
return &Controller{impl: impl}
}
type Controller struct {
impl s3.Interface
}
func (c *Controller) HashPath(md5 string) string {
return path.Join(hashPath, md5)
}
func (c *Controller) NowPath() string {
now := time.Now()
return path.Join(
fmt.Sprintf("%04d", now.Year()),
fmt.Sprintf("%02d", now.Month()),
fmt.Sprintf("%02d", now.Day()),
fmt.Sprintf("%02d", now.Hour()),
fmt.Sprintf("%02d", now.Minute()),
fmt.Sprintf("%02d", now.Second()),
)
}
func (c *Controller) UUID() string {
id := uuid.New()
return hex.EncodeToString(id[:])
}
func (c *Controller) PartSize(ctx context.Context, size int64) (int64, error) {
return c.impl.PartSize(ctx, size)
}
func (c *Controller) PartLimit() *s3.PartLimit {
return c.impl.PartLimit()
}
func (c *Controller) GetHashObject(ctx context.Context, hash string) (*s3.ObjectInfo, error) {
return c.impl.StatObject(ctx, c.HashPath(hash))
}
func (c *Controller) InitiateUpload(
ctx context.Context,
hash string,
size int64,
expire time.Duration,
maxParts int,
) (*InitiateUploadResult, error) {
merge v3dev into main (#504) * statistics user register * refactor: router change * minio init * UserRegisterCount * push use local conn * refactor: user pb update * remove online push close grpc conn * refactor: user pb update * refactor:pb file * msgs statistics * msgs statistics * revoke userID * refactor: errcode update * active user * active user * active user * refactor: errcode update * feat: conn update token * active user * active user * feat: conn update token * active user * feat: conn update token * feat: conn update token * feat: conn update token * add tx_oss cos * active user * active user * group create * group create * feat: group notification show to conversation * feat: group notification show to conversation * group active * user active * sendNotificationWithName * withname * privateChat * a2r call option * grpc with detail return error * change log error * chain unary interceptor * api nil slice map * fix sync has read * fix: text update * fix: update add model * set conversations update * set privateChat * fix: content update * remove unuse rpc * msgDestruct * cron use rpc mw * set IsMsgDestruct * msg destruct * msgDestruct * s3 minio, cos, oss support * feat: add implement of GetUsersOnlineStatus, #472 (#477) * s3 minio, cos, oss support * s3 route * remove extendMsg code * s3 route * remove unuse code * s3 pb * s3 pb * s3 pb * s3 presigned put * s3 presigned test * s3 presigned test * s3 presigned test * s3 presigned test * s3 presigned test * s3 presigned test * s3 presigned test * s3 presigned test * Update .gitignore (#482) * s3 debug log * s3 debug log * cron add log and fix cron * add log * cron * s3 config * fix kick user bug * s3 cos * add kick log * s3 cos test * s3 cos test * s3 cos test * kick user log * kickuserlog * s3 cos copy * s3 cos copy * s3 url * s3 url * s3 AccessURL * log * s3 InitiateMultipartUpload add ExpireTime --------- Co-authored-by: withchao <993506633@qq.com> Co-authored-by: wangchuxiao <wangchuxiao97@outlook.com> Co-authored-by: BanTanger <88583317+BanTanger@users.noreply.github.com> Co-authored-by: withchao <48119764+withchao@users.noreply.github.com> Co-authored-by: Alan <68671759+hanzhixiao@users.noreply.github.com>
1 year ago
defer log.ZDebug(ctx, "return")
if size < 0 {
return nil, errors.New("invalid size")
}
if hashBytes, err := hex.DecodeString(hash); err != nil {
return nil, err
} else if len(hashBytes) != md5.Size {
return nil, errors.New("invalid md5")
}
partSize, err := c.impl.PartSize(ctx, size)
if err != nil {
return nil, err
}
partNumber := int(size / partSize)
if size%partSize > 0 {
partNumber++
}
if maxParts > 0 && partNumber > 0 && partNumber < maxParts {
return nil, errors.New(fmt.Sprintf("too many parts: %d", partNumber))
}
if info, err := c.impl.StatObject(ctx, c.HashPath(hash)); err == nil {
return nil, &HashAlreadyExistsError{Object: info}
} else if !c.impl.IsNotFound(err) {
return nil, err
}
if size <= partSize {
// 预签名上传
key := path.Join(tempPath, c.NowPath(), fmt.Sprintf("%s_%d_%s.presigned", hash, size, c.UUID()))
rawURL, err := c.impl.PresignedPutObject(ctx, key, expire)
if err != nil {
return nil, err
}
return &InitiateUploadResult{
UploadID: newMultipartUploadID(multipartUploadID{
Type: UploadTypePresigned,
ID: "",
Key: key,
Size: size,
Hash: hash,
}),
PartSize: partSize,
Sign: &s3.AuthSignResult{
Parts: []s3.SignPart{
{
PartNumber: 1,
URL: rawURL,
},
},
},
}, nil
} else {
// 分片上传
upload, err := c.impl.InitiateMultipartUpload(ctx, c.HashPath(hash))
if err != nil {
return nil, err
}
if maxParts < 0 {
maxParts = partNumber
}
var authSign *s3.AuthSignResult
if maxParts > 0 {
partNumbers := make([]int, partNumber)
for i := 0; i < maxParts; i++ {
partNumbers[i] = i + 1
}
authSign, err = c.impl.AuthSign(ctx, upload.UploadID, upload.Key, time.Hour*24, partNumbers)
if err != nil {
return nil, err
}
}
return &InitiateUploadResult{
UploadID: newMultipartUploadID(multipartUploadID{
Type: UploadTypeMultipart,
ID: upload.UploadID,
Key: upload.Key,
Size: size,
Hash: hash,
}),
PartSize: partSize,
Sign: authSign,
}, nil
}
}
func (c *Controller) CompleteUpload(ctx context.Context, uploadID string, partHashs []string) (*UploadResult, error) {
defer log.ZDebug(ctx, "return")
upload, err := parseMultipartUploadID(uploadID)
if err != nil {
return nil, err
}
if md5Sum := md5.Sum([]byte(strings.Join(partHashs, partSeparator))); hex.EncodeToString(md5Sum[:]) != upload.Hash {
fmt.Println("CompleteUpload sum:", hex.EncodeToString(md5Sum[:]), "upload hash:", upload.Hash)
return nil, errors.New("md5 mismatching")
}
if info, err := c.impl.StatObject(ctx, c.HashPath(upload.Hash)); err == nil {
return &UploadResult{
Key: info.Key,
Size: info.Size,
Hash: info.ETag,
}, nil
} else if !c.impl.IsNotFound(err) {
return nil, err
}
cleanObject := make(map[string]struct{})
defer func() {
for key := range cleanObject {
_ = c.impl.DeleteObject(ctx, key)
}
}()
var targetKey string
switch upload.Type {
case UploadTypeMultipart:
parts := make([]s3.Part, len(partHashs))
for i, part := range partHashs {
parts[i] = s3.Part{
PartNumber: i + 1,
ETag: part,
}
}
// todo: 验证大小
result, err := c.impl.CompleteMultipartUpload(ctx, upload.ID, upload.Key, parts)
if err != nil {
return nil, err
}
targetKey = result.Key
case UploadTypePresigned:
uploadInfo, err := c.impl.StatObject(ctx, upload.Key)
if err != nil {
return nil, err
}
cleanObject[uploadInfo.Key] = struct{}{}
if uploadInfo.Size != upload.Size {
return nil, errors.New("upload size mismatching")
}
md5Sum := md5.Sum([]byte(strings.Join([]string{uploadInfo.ETag}, partSeparator)))
if md5val := hex.EncodeToString(md5Sum[:]); md5val != upload.Hash {
return nil, errs.ErrArgs.Wrap(fmt.Sprintf("md5 mismatching %s != %s", md5val, upload.Hash))
}
// 防止在这个时候,并发操作,导致文件被覆盖
copyInfo, err := c.impl.CopyObject(ctx, uploadInfo.Key, upload.Key+"."+c.UUID())
if err != nil {
return nil, err
}
cleanObject[copyInfo.Key] = struct{}{}
if copyInfo.ETag != uploadInfo.ETag {
return nil, errors.New("[concurrency]copy md5 mismatching")
}
hashCopyInfo, err := c.impl.CopyObject(ctx, copyInfo.Key, c.HashPath(upload.Hash))
if err != nil {
return nil, err
}
log.ZInfo(ctx, "hashCopyInfo", "value", fmt.Sprintf("%+v", hashCopyInfo))
targetKey = hashCopyInfo.Key
default:
return nil, errors.New("invalid upload id type")
}
return &UploadResult{
Key: targetKey,
Size: upload.Size,
Hash: upload.Hash,
}, nil
}
func (c *Controller) AuthSign(ctx context.Context, uploadID string, partNumbers []int) (*s3.AuthSignResult, error) {
upload, err := parseMultipartUploadID(uploadID)
if err != nil {
return nil, err
}
switch upload.Type {
case UploadTypeMultipart:
return c.impl.AuthSign(ctx, upload.ID, upload.Key, time.Hour*24, partNumbers)
case UploadTypePresigned:
return nil, errors.New("presigned id not support auth sign")
default:
return nil, errors.New("invalid upload id type")
}
}
func (c *Controller) IsNotFound(err error) bool {
return c.impl.IsNotFound(err)
}
func (c *Controller) AccessURL(
ctx context.Context,
name string,
expire time.Duration,
opt *s3.AccessURLOption,
) (string, error) {
merge v3dev into main (#504) * statistics user register * refactor: router change * minio init * UserRegisterCount * push use local conn * refactor: user pb update * remove online push close grpc conn * refactor: user pb update * refactor:pb file * msgs statistics * msgs statistics * revoke userID * refactor: errcode update * active user * active user * active user * refactor: errcode update * feat: conn update token * active user * active user * feat: conn update token * active user * feat: conn update token * feat: conn update token * feat: conn update token * add tx_oss cos * active user * active user * group create * group create * feat: group notification show to conversation * feat: group notification show to conversation * group active * user active * sendNotificationWithName * withname * privateChat * a2r call option * grpc with detail return error * change log error * chain unary interceptor * api nil slice map * fix sync has read * fix: text update * fix: update add model * set conversations update * set privateChat * fix: content update * remove unuse rpc * msgDestruct * cron use rpc mw * set IsMsgDestruct * msg destruct * msgDestruct * s3 minio, cos, oss support * feat: add implement of GetUsersOnlineStatus, #472 (#477) * s3 minio, cos, oss support * s3 route * remove extendMsg code * s3 route * remove unuse code * s3 pb * s3 pb * s3 pb * s3 presigned put * s3 presigned test * s3 presigned test * s3 presigned test * s3 presigned test * s3 presigned test * s3 presigned test * s3 presigned test * s3 presigned test * Update .gitignore (#482) * s3 debug log * s3 debug log * cron add log and fix cron * add log * cron * s3 config * fix kick user bug * s3 cos * add kick log * s3 cos test * s3 cos test * s3 cos test * kick user log * kickuserlog * s3 cos copy * s3 cos copy * s3 url * s3 url * s3 AccessURL * log * s3 InitiateMultipartUpload add ExpireTime --------- Co-authored-by: withchao <993506633@qq.com> Co-authored-by: wangchuxiao <wangchuxiao97@outlook.com> Co-authored-by: BanTanger <88583317+BanTanger@users.noreply.github.com> Co-authored-by: withchao <48119764+withchao@users.noreply.github.com> Co-authored-by: Alan <68671759+hanzhixiao@users.noreply.github.com>
1 year ago
return c.impl.AccessURL(ctx, name, expire, opt)
}