From 00f0d9fb424cd5b95b7396703a2ac64e1e562cac Mon Sep 17 00:00:00 2001 From: withchao <993506633@qq.com> Date: Fri, 15 Dec 2023 19:07:21 +0800 Subject: [PATCH] s3 form data --- go.mod | 4 +- go.sum | 2 - internal/api/route.go | 2 + internal/api/third.go | 8 ++ internal/rpc/third/s3.go | 112 ++++++++++++++++++++++ internal/rpc/third/tool.go | 3 + pkg/common/cmd/root.go | 2 +- pkg/common/db/controller/s3.go | 10 ++ pkg/common/db/s3/cont/consts.go | 1 + pkg/common/db/s3/cont/controller.go | 4 + pkg/common/db/s3/cos/cos.go | 4 + pkg/common/db/s3/minio/minio.go | 36 +++++++ pkg/common/db/s3/minio/minio_test.go | 37 +++++++ pkg/common/db/s3/oss/oss.go | 4 + pkg/common/db/s3/s3.go | 10 ++ pkg/common/ginprometheus/ginprometheus.go | 2 +- 16 files changed, 236 insertions(+), 5 deletions(-) create mode 100644 pkg/common/db/s3/minio/minio_test.go diff --git a/go.mod b/go.mod index 11d374d08..4a0520e67 100644 --- a/go.mod +++ b/go.mod @@ -4,7 +4,7 @@ go 1.19 require ( firebase.google.com/go v3.13.0+incompatible - github.com/OpenIMSDK/protocol v0.0.31 + github.com/OpenIMSDK/protocol v0.0.35 github.com/OpenIMSDK/tools v0.0.20 github.com/bwmarrin/snowflake v0.3.0 // indirect github.com/dtm-labs/rockscache v0.1.1 @@ -154,3 +154,5 @@ require ( golang.org/x/crypto v0.14.0 // indirect gopkg.in/ini.v1 v1.67.0 // indirect ) + +replace github.com/OpenIMSDK/protocol => C:\Users\openIM\Desktop\fork\protocol diff --git a/go.sum b/go.sum index 30e4b3cb4..9684c1079 100644 --- a/go.sum +++ b/go.sum @@ -18,8 +18,6 @@ firebase.google.com/go v3.13.0+incompatible/go.mod h1:xlah6XbEyW6tbfSklcfe5FHJIw github.com/BurntSushi/toml v0.3.1/go.mod h1:xHWCNGjB5oqiDr8zfno3MHue2Ht5sIBksp03qcyfWMU= github.com/IBM/sarama v1.41.3 h1:MWBEJ12vHC8coMjdEXFq/6ftO6DUZnQlFYcxtOJFa7c= github.com/IBM/sarama v1.41.3/go.mod h1:Xxho9HkHd4K/MDUo/T/sOqwtX/17D33++E9Wib6hUdQ= -github.com/OpenIMSDK/protocol v0.0.31 h1:ax43x9aqA6EKNXNukS5MT5BSTqkUmwO4uTvbJLtzCgE= -github.com/OpenIMSDK/protocol v0.0.31/go.mod h1:F25dFrwrIx3lkNoiuf6FkCfxuwf8L4Z8UIsdTHP/r0Y= github.com/OpenIMSDK/tools v0.0.20 h1:zBTjQZRJ5lR1FIzP9mtWyAvh5dKsmJXQugi4p8X/97k= github.com/OpenIMSDK/tools v0.0.20/go.mod h1:eg+q4A34Qmu73xkY0mt37FHGMCMfC6CtmOnm0kFEGFI= github.com/QcloudApi/qcloud_sign_golang v0.0.0-20141224014652-e4130a326409/go.mod h1:1pk82RBxDY/JZnPQrtqHlUFfCctgdorsd9M06fMynOM= diff --git a/internal/api/route.go b/internal/api/route.go index 7a331d643..0b8be95fd 100644 --- a/internal/api/route.go +++ b/internal/api/route.go @@ -161,6 +161,8 @@ func NewGinRouter(discov discoveryregistry.SvcDiscoveryRegistry, rdb redis.Unive objectGroup.POST("/auth_sign", t.AuthSign) objectGroup.POST("/complete_multipart_upload", t.CompleteMultipartUpload) objectGroup.POST("/access_url", t.AccessURL) + objectGroup.POST("/initiate_form_data", t.InitiateFormData) + objectGroup.POST("/complete_form_data", t.CompleteFormData) objectGroup.GET("/*name", t.ObjectRedirect) } // Message diff --git a/internal/api/third.go b/internal/api/third.go index 5191903da..37ec55098 100644 --- a/internal/api/third.go +++ b/internal/api/third.go @@ -71,6 +71,14 @@ func (o *ThirdApi) AccessURL(c *gin.Context) { a2r.Call(third.ThirdClient.AccessURL, o.Client, c) } +func (o *ThirdApi) InitiateFormData(c *gin.Context) { + a2r.Call(third.ThirdClient.InitiateFormData, o.Client, c) +} + +func (o *ThirdApi) CompleteFormData(c *gin.Context) { + a2r.Call(third.ThirdClient.CompleteFormData, o.Client, c) +} + func (o *ThirdApi) ObjectRedirect(c *gin.Context) { name := c.Param("name") if name == "" { diff --git a/internal/rpc/third/s3.go b/internal/rpc/third/s3.go index ca826e805..14d1d5176 100644 --- a/internal/rpc/third/s3.go +++ b/internal/rpc/third/s3.go @@ -16,6 +16,12 @@ package third import ( "context" + "encoding/base64" + "encoding/hex" + "encoding/json" + "github.com/google/uuid" + "github.com/openimsdk/open-im-server/v3/pkg/authverify" + "path" "strconv" "time" @@ -179,6 +185,112 @@ func (t *thirdServer) AccessURL(ctx context.Context, req *third.AccessURLReq) (* }, nil } +const direct = "direct" + +func (t *thirdServer) InitiateFormData(ctx context.Context, req *third.InitiateFormDataReq) (*third.InitiateFormDataResp, error) { + if req.Name == "" { + return nil, errs.ErrArgs.Wrap("name is empty") + } + if req.Size <= 0 { + return nil, errs.ErrArgs.Wrap("size must be greater than 0") + } + if err := checkUploadName(ctx, req.Name); err != nil { + return nil, err + } + var duration time.Duration + opUserID := mcontext.GetOpUserID(ctx) + var key string + if authverify.IsManagerUserID(opUserID) { + if req.Millisecond <= 0 { + duration = time.Minute * 10 + } else { + duration = time.Millisecond * time.Duration(req.Millisecond) + } + if req.Absolute { + key = req.Name + } + } else { + duration = time.Minute * 10 + } + uid, err := uuid.NewRandom() + if err != nil { + return nil, err + } + if key == "" { + date := time.Now().Format("20060102") + key = path.Join(cont.DirectPath, date, opUserID, hex.EncodeToString(uid[:])+path.Ext(req.Name)) + } + mate := FormDataMate{ + Name: req.Name, + Size: req.Size, + ContentType: req.ContentType, + Group: req.Group, + Key: key, + } + mateData, err := json.Marshal(&mate) + if err != nil { + return nil, err + } + resp, err := t.s3dataBase.FormData(ctx, key, req.Size, req.ContentType, duration) + if err != nil { + return nil, err + } + return &third.InitiateFormDataResp{ + Id: base64.RawStdEncoding.EncodeToString(mateData), + Url: resp.URL, + File: resp.File, + Header: toPbMapArray(resp.Header), + FormData: resp.FormData, + Expires: resp.Expires.UnixMilli(), + }, nil +} + +func (t *thirdServer) CompleteFormData(ctx context.Context, req *third.CompleteFormDataReq) (*third.CompleteFormDataResp, error) { + if req.Id == "" { + return nil, errs.ErrArgs.Wrap("id is empty") + } + data, err := base64.RawStdEncoding.DecodeString(req.Id) + if err != nil { + return nil, errs.ErrArgs.Wrap("invalid id " + err.Error()) + } + var mate FormDataMate + if err := json.Unmarshal(data, &mate); err != nil { + return nil, errs.ErrArgs.Wrap("invalid id " + err.Error()) + } + if err := checkUploadName(ctx, mate.Name); err != nil { + return nil, err + } + info, err := t.s3dataBase.StatObject(ctx, mate.Key) + if err != nil { + return nil, err + } + if info.Size > 0 && info.Size != mate.Size { + return nil, errs.ErrData.Wrap("file size mismatch") + } + obj := &relation.ObjectModel{ + Name: mate.Name, + UserID: mcontext.GetOpUserID(ctx), + Hash: "etag_" + info.ETag, + Key: info.Key, + Size: info.Size, + ContentType: mate.ContentType, + Group: mate.Group, + CreateTime: time.Now(), + } + if err := t.s3dataBase.SetObject(ctx, obj); err != nil { + return nil, err + } + return &third.CompleteFormDataResp{Url: t.apiAddress(mate.Name)}, nil +} + func (t *thirdServer) apiAddress(name string) string { return t.apiURL + name } + +type FormDataMate struct { + Name string `json:"name"` + Size int64 `json:"size"` + ContentType string `json:"contentType"` + Group string `json:"group"` + Key string `json:"key"` +} diff --git a/internal/rpc/third/tool.go b/internal/rpc/third/tool.go index a65d882dd..a6c16ff9d 100644 --- a/internal/rpc/third/tool.go +++ b/internal/rpc/third/tool.go @@ -29,6 +29,9 @@ import ( ) func toPbMapArray(m map[string][]string) []*third.KeyValues { + if len(m) == 0 { + return nil + } res := make([]*third.KeyValues, 0, len(m)) for key := range m { res = append(res, &third.KeyValues{ diff --git a/pkg/common/cmd/root.go b/pkg/common/cmd/root.go index 0bc308e07..66bec61a7 100644 --- a/pkg/common/cmd/root.go +++ b/pkg/common/cmd/root.go @@ -45,7 +45,7 @@ type CmdOpts struct { func WithCronTaskLogName() func(*CmdOpts) { return func(opts *CmdOpts) { - opts.loggerPrefixName = "OpenIM.CronTask.log.all" + opts.loggerPrefixName = "openim.crontask.log.all" } } diff --git a/pkg/common/db/controller/s3.go b/pkg/common/db/controller/s3.go index 6916a7d30..95505de41 100644 --- a/pkg/common/db/controller/s3.go +++ b/pkg/common/db/controller/s3.go @@ -35,6 +35,8 @@ type S3Database interface { CompleteMultipartUpload(ctx context.Context, uploadID string, parts []string) (*cont.UploadResult, error) AccessURL(ctx context.Context, name string, expire time.Duration, opt *s3.AccessURLOption) (time.Time, string, error) SetObject(ctx context.Context, info *relation.ObjectModel) error + StatObject(ctx context.Context, name string) (*s3.ObjectInfo, error) + FormData(ctx context.Context, name string, size int64, contentType string, duration time.Duration) (*s3.FormData, error) } func NewS3Database(rdb redis.UniversalClient, s3 s3.Interface, obj relation.ObjectInfoModelInterface) S3Database { @@ -100,3 +102,11 @@ func (s *s3Database) AccessURL(ctx context.Context, name string, expire time.Dur } return expireTime, rawURL, nil } + +func (s *s3Database) StatObject(ctx context.Context, name string) (*s3.ObjectInfo, error) { + return s.s3.StatObject(ctx, name) +} + +func (s *s3Database) FormData(ctx context.Context, name string, size int64, contentType string, duration time.Duration) (*s3.FormData, error) { + return s.s3.FormData(ctx, name, size, contentType, duration) +} diff --git a/pkg/common/db/s3/cont/consts.go b/pkg/common/db/s3/cont/consts.go index 1a0467ce5..a01a8312c 100644 --- a/pkg/common/db/s3/cont/consts.go +++ b/pkg/common/db/s3/cont/consts.go @@ -17,6 +17,7 @@ package cont const ( hashPath = "openim/data/hash/" tempPath = "openim/temp/" + DirectPath = "openim/direct" UploadTypeMultipart = 1 // 分片上传 UploadTypePresigned = 2 // 预签名上传 partSeparator = "," diff --git a/pkg/common/db/s3/cont/controller.go b/pkg/common/db/s3/cont/controller.go index 1bf1a4b12..82c27c1f2 100644 --- a/pkg/common/db/s3/cont/controller.go +++ b/pkg/common/db/s3/cont/controller.go @@ -279,3 +279,7 @@ func (c *Controller) AccessURL(ctx context.Context, name string, expire time.Dur } return c.impl.AccessURL(ctx, name, expire, opt) } + +func (c *Controller) FormData(ctx context.Context, name string, size int64, contentType string, duration time.Duration) (*s3.FormData, error) { + return c.impl.FormData(ctx, name, size, contentType, duration) +} diff --git a/pkg/common/db/s3/cos/cos.go b/pkg/common/db/s3/cos/cos.go index 7add88487..ca8eb1694 100644 --- a/pkg/common/db/s3/cos/cos.go +++ b/pkg/common/db/s3/cos/cos.go @@ -326,3 +326,7 @@ func (c *Cos) getPresignedURL(ctx context.Context, name string, expire time.Dura } return c.client.Object.GetObjectURL(name), nil } + +func (c *Cos) FormData(ctx context.Context, name string, size int64, contentType string, duration time.Duration) (*s3.FormData, error) { + return nil, errors.New("cos temporarily not supported") +} diff --git a/pkg/common/db/s3/minio/minio.go b/pkg/common/db/s3/minio/minio.go index be49e2faa..914e2ae54 100644 --- a/pkg/common/db/s3/minio/minio.go +++ b/pkg/common/db/s3/minio/minio.go @@ -441,3 +441,39 @@ func (m *Minio) getObjectData(ctx context.Context, name string, limit int64) ([] } return io.ReadAll(io.LimitReader(object, limit)) } + +func (m *Minio) FormData(ctx context.Context, name string, size int64, contentType string, duration time.Duration) (*s3.FormData, error) { + if err := m.initMinio(ctx); err != nil { + return nil, err + } + policy := minio.NewPostPolicy() + if err := policy.SetKey(name); err != nil { + return nil, err + } + expires := time.Now().Add(duration) + if err := policy.SetExpires(expires); err != nil { + return nil, err + } + if err := policy.SetContentLengthRange(0, size); err != nil { + return nil, err + } + if contentType != "" { + if err := policy.SetContentType(contentType); err != nil { + return nil, err + } + } + if err := policy.SetBucket(config.Config.Object.Minio.Bucket); err != nil { + return nil, err + } + u, fd, err := m.core.PresignedPostPolicy(ctx, policy) + if err != nil { + panic(err) + } + return &s3.FormData{ + URL: u.String(), + File: "file", + Header: nil, + FormData: fd, + Expires: expires, + }, nil +} diff --git a/pkg/common/db/s3/minio/minio_test.go b/pkg/common/db/s3/minio/minio_test.go new file mode 100644 index 000000000..2c721eb25 --- /dev/null +++ b/pkg/common/db/s3/minio/minio_test.go @@ -0,0 +1,37 @@ +package minio + +import ( + "context" + "github.com/minio/minio-go/v7" + "github.com/openimsdk/open-im-server/v3/pkg/common/config" + "testing" + "time" +) + +func TestName(t *testing.T) { + config.Config.Object.Minio.Bucket = "openim" + config.Config.Object.Minio.AccessKeyID = "root" + config.Config.Object.Minio.SecretAccessKey = "openIM123" + config.Config.Object.Minio.Endpoint = "http://172.16.8.38:10005" + tmp, err := NewMinio(nil) + if err != nil { + panic(err) + } + min := tmp.(*Minio) + cli := min.core.Client + ctx, cancel := context.WithTimeout(context.Background(), time.Second*3) + defer cancel() + policy := minio.NewPostPolicy() + _ = policy.SetExpires(time.Now().Add(time.Hour)) + _ = policy.SetKey("test.txt") + _ = policy.SetBucket(config.Config.Object.Minio.Bucket) + policy.SetContentType("text/plain") + u, fd, err := cli.PresignedPostPolicy(ctx, policy) + if err != nil { + panic(err) + } + t.Log(u) + for k, v := range fd { + t.Log(k, v) + } +} diff --git a/pkg/common/db/s3/oss/oss.go b/pkg/common/db/s3/oss/oss.go index 6a728127b..a4c34fa82 100644 --- a/pkg/common/db/s3/oss/oss.go +++ b/pkg/common/db/s3/oss/oss.go @@ -327,3 +327,7 @@ func (o *OSS) AccessURL(ctx context.Context, name string, expire time.Duration, params := getURLParams(*o.bucket.Client.Conn, rawParams) return getURL(o.um, o.bucket.BucketName, name, params).String(), nil } + +func (o *OSS) FormData(ctx context.Context, name string, size int64, contentType string, duration time.Duration) (*s3.FormData, error) { + return nil, errors.New("oss temporarily not supported") +} diff --git a/pkg/common/db/s3/s3.go b/pkg/common/db/s3/s3.go index afbe91955..3ca29943b 100644 --- a/pkg/common/db/s3/s3.go +++ b/pkg/common/db/s3/s3.go @@ -74,6 +74,14 @@ type CopyObjectInfo struct { ETag string `json:"etag"` } +type FormData struct { + URL string `json:"url"` + File string `json:"file"` + Header http.Header `json:"header"` + FormData map[string]string `json:"form"` + Expires time.Time `json:"expires"` +} + type SignPart struct { PartNumber int `json:"partNumber"` URL string `json:"url"` @@ -152,4 +160,6 @@ type Interface interface { ListUploadedParts(ctx context.Context, uploadID string, name string, partNumberMarker int, maxParts int) (*ListUploadedPartsResult, error) AccessURL(ctx context.Context, name string, expire time.Duration, opt *AccessURLOption) (string, error) + + FormData(ctx context.Context, name string, size int64, contentType string, duration time.Duration) (*FormData, error) } diff --git a/pkg/common/ginprometheus/ginprometheus.go b/pkg/common/ginprometheus/ginprometheus.go index a325595d6..132e97770 100644 --- a/pkg/common/ginprometheus/ginprometheus.go +++ b/pkg/common/ginprometheus/ginprometheus.go @@ -418,7 +418,7 @@ func computeApproximateRequestSize(r *http.Request) int { } s += len(r.Host) - // r.Form and r.MultipartForm are assumed to be included in r.URL. + // r.FormData and r.MultipartForm are assumed to be included in r.URL. if r.ContentLength != -1 { s += int(r.ContentLength)