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/v2/analysis/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 "" }