105 lines
2.8 KiB
Go
105 lines
2.8 KiB
Go
package analysisdevicelogic
|
|
|
|
import (
|
|
"context"
|
|
|
|
"lone-services/pkg/utils"
|
|
analysis "lone-services/rpc/analysis/pb"
|
|
"lone-services/services/analysis/internal/callback"
|
|
analysisoss "lone-services/services/analysis/internal/oss"
|
|
"lone-services/services/analysis/internal/svc"
|
|
|
|
"github.com/zeromicro/go-zero/core/logx"
|
|
"google.golang.org/grpc/codes"
|
|
"google.golang.org/grpc/metadata"
|
|
"google.golang.org/grpc/status"
|
|
)
|
|
|
|
const ossCallbackPath = "/api/v1/oss/callback"
|
|
|
|
type OssCallbackLogic struct {
|
|
ctx context.Context
|
|
svcCtx *svc.ServiceContext
|
|
logx.Logger
|
|
}
|
|
|
|
func NewOssCallbackLogic(ctx context.Context, svcCtx *svc.ServiceContext) *OssCallbackLogic {
|
|
return &OssCallbackLogic{
|
|
ctx: ctx,
|
|
svcCtx: svcCtx,
|
|
Logger: logx.WithContext(ctx),
|
|
}
|
|
}
|
|
|
|
func (l *OssCallbackLogic) OssCallback(in *analysis.OssCallbackReq) (*analysis.OssCallbackResp, error) {
|
|
rawBody := utils.GetBody(l.ctx)
|
|
if rawBody == utils.StringEmpty {
|
|
l.Error("oss callback empty raw body")
|
|
return nil, status.Error(codes.InvalidArgument, "read body failed")
|
|
}
|
|
|
|
path, query, auth, pubKeyURL := callbackMeta(l.ctx)
|
|
if path == utils.StringEmpty {
|
|
path = ossCallbackPath
|
|
}
|
|
|
|
if err := analysisoss.VerifyCallback(analysisoss.VerifyInput{
|
|
Path: path,
|
|
RawQuery: query,
|
|
Body: []byte(rawBody),
|
|
Authorization: auth,
|
|
PubKeyURLHdr: pubKeyURL,
|
|
}); err != nil {
|
|
l.Errorf("oss callback verify failed: %v", err)
|
|
return nil, status.Error(codes.PermissionDenied, "verify not ok")
|
|
}
|
|
|
|
payload := analysisoss.CallbackPayload{
|
|
Bucket: in.GetBucket(),
|
|
Object: in.GetObject(),
|
|
Etag: in.GetEtag(),
|
|
Size: in.GetSize(),
|
|
MimeType: in.GetMimeType(),
|
|
TaskId: in.GetTaskId(),
|
|
}
|
|
if payload.TaskId == utils.StringEmpty || payload.Object == utils.StringEmpty {
|
|
l.Error("oss callback missing task_id or object")
|
|
return nil, status.Error(codes.InvalidArgument, "invalid body")
|
|
}
|
|
|
|
go func(p analysisoss.CallbackPayload) {
|
|
defer func() {
|
|
if r := recover(); r != nil {
|
|
logx.Errorf("oss callback process panic: %v", r)
|
|
}
|
|
}()
|
|
if err := callback.ProcessUpload(context.Background(), p); err != nil {
|
|
logx.Errorf("oss callback process failed: %v", err)
|
|
}
|
|
}(payload)
|
|
|
|
return &analysis.OssCallbackResp{Status: "OK"}, nil
|
|
}
|
|
|
|
func callbackMeta(ctx context.Context) (path, query, auth, pubKeyURL string) {
|
|
md, ok := metadata.FromIncomingContext(ctx)
|
|
if !ok {
|
|
return
|
|
}
|
|
path = firstMD(md, "x-request-path", "x-original-path")
|
|
query = firstMD(md, "x-request-query")
|
|
auth = firstMD(md, "x-oss-authorization", "authorization", "grpcgateway-authorization")
|
|
pubKeyURL = firstMD(md, "x-oss-pub-key-url", "grpcgateway-x-oss-pub-key-url")
|
|
return
|
|
}
|
|
|
|
func firstMD(md metadata.MD, keys ...string) string {
|
|
for _, key := range keys {
|
|
vals := md.Get(key)
|
|
if len(vals) > 0 && vals[0] != "" {
|
|
return vals[0]
|
|
}
|
|
}
|
|
return ""
|
|
}
|