lone-services/services/analysis/internal/logic/analysisdevice/ossCallbackLogic.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/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 ""
}