package server

import (
	"context"
	"errors"
	"fmt"
	"io"
	"net/http"
	"net/url"
	"path"
	"strings"
	"time"

	"github.com/Azure/azure-sdk-for-go/sdk/azcore"
	"github.com/Azure/azure-sdk-for-go/sdk/storage/azblob"
	"github.com/aliyun/alibabacloud-oss-go-sdk-v2/oss"
	osscredentials "github.com/aliyun/alibabacloud-oss-go-sdk-v2/oss/credentials"
	"github.com/aws/aws-sdk-go-v2/aws"
	"github.com/aws/aws-sdk-go-v2/config"
	"github.com/aws/aws-sdk-go-v2/credentials"
	"github.com/aws/aws-sdk-go-v2/service/s3"
	"github.com/aws/smithy-go"
	"github.com/gogo/protobuf/proto"
	kvbackup "github.com/pingcap/kvproto/pkg/brpb"
	apiv1 "github.com/tidbcloud/cluster-service-ng/api/cluster-service-ng/v1beta"
	"github.com/tidbcloud/cluster-service-ng/pkg/utils"
	"github.com/tidbcloud/tiinf/monitor/tilog"
	"go.uber.org/zap"
	"google.golang.org/grpc/codes"
	"google.golang.org/grpc/status"
	"google.golang.org/protobuf/types/known/timestamppb"
)

type TestClassicBackupAccessInput struct {
	*apiv1.TestClassicBackupAccessRequest
}

type TestClassicBackupAccessOutput struct {
	*apiv1.TestClassicBackupAccessResponse
}

func (s *serverImpl) TestClassicBackupAccess(ctx context.Context, input *TestClassicBackupAccessInput) (*TestClassicBackupAccessOutput, error) {
	switch input.CloudProvider {
	case "aws":
		return s.testClassicBackupAccessAWS(ctx, input)
	case "alicloud":
		return s.testClassicBackupAccessAlicloud(ctx, input)
	case "azure":
		return s.testClassicBackupAccessAzure(ctx, input)
	default:
		return nil, status.Errorf(codes.InvalidArgument, "invalid cloud provider: %s", input.CloudProvider)
	}
}

const (
	NoSuchBucket          = "NoSuchBucket"
	NoSuchKey             = "NoSuchKey"
	AccessDenied          = "AccessDenied"
	InvalidAccessKeyId    = "InvalidAccessKeyId"
	SignatureDoesNotMatch = "SignatureDoesNotMatch"

	BackupMetaName = "backupmeta"
)

/*
aws s3 policy example:

	{
	    "Version": "2012-10-17",
	    "Statement": [
	        {
	            "Sid": "AllowGetBucketLocation",
	            "Effect": "Allow",
	            "Action": "s3:GetBucketLocation",
	            "Resource": "arn:aws:s3:::caiyuanrui-us-west-2"
	        },
	        {
	            "Sid": "AllowListPrefix",
	            "Effect": "Allow",
	            "Action": "s3:ListBucket",
	            "Resource": "arn:aws:s3:::caiyuanrui-us-west-2",
	            "Condition": {
	                "StringLike": {
	                    "s3:prefix": "test-backup/testbr1-2861990/*"
	                }
	            }
	        },
	        {
	            "Sid": "AllowReadObjectsInPrefix",
	            "Effect": "Allow",
	            "Action": "s3:GetObject",
	            "Resource": "arn:aws:s3:::caiyuanrui-us-west-2/test-backup/testbr1-2861990/*"
	        }
	    ]
	}
*/
func (s *serverImpl) testClassicBackupAccessAWS(ctx context.Context, input *TestClassicBackupAccessInput) (*TestClassicBackupAccessOutput, error) {
	u, err := url.Parse(input.BucketUri)
	if err != nil {
		tilog.Warn(ctx, "invalid bucket uri", zap.Error(err), zap.String("bucket_uri", utils.RedactURL(input.BucketUri)))
		return nil, status.Error(codes.InvalidArgument, "invalid bucket uri")
	}
	bucketName := u.Hostname()
	backupPrefix := u.Path

	// include bucket existing test
	client, realRegion, resp, err := awsBuildBucketClient(ctx, bucketName, input.AccessKeyId, input.AccessKeySecret)
	if client == nil {
		tilog.Warn(ctx, "failed to build bucket client", zap.Error(err), zap.Any("resp", resp))
		return resp, err
	}

	// test list objects
	resp, err = awsTestListObjects(ctx, bucketName, backupPrefix, client)
	if err != nil || resp != nil {
		tilog.Warn(ctx, "failed to test list objects", zap.Error(err), zap.Any("resp", resp))
		return resp, err
	}

	// test get object and backup meta
	backupMeta, resp, err := awsGetBackupMeta(ctx, bucketName, backupPrefix, client)
	if backupMeta == nil {
		tilog.Warn(ctx, "failed to get backup meta", zap.Error(err), zap.Any("resp", resp))
		return resp, err
	}

	resp = makeTestClassicBackupAccessOutput(backupMeta)
	resp.Region = realRegion
	tilog.Info(ctx, "test classic backup access succeeded", zap.Any("resp", resp))
	return resp, nil
}

/*
alicloud oss policy example:

	{
	  "Version": "1",
	  "Statement": [
	    {
	      "Effect": "Allow",
	      "Action": "oss:ListObjects",
	      "Resource": "acs:oss:*:*:caiyuanrui-ap-southeast-1",
	      "Condition": {
	        "StringLike": {
	          "oss:Prefix": "test-backup/testbr1-2861990/*"
	        }
	      }
	    },
	    {
	      "Effect": "Allow",
	      "Action": "oss:GetObject",
	      "Resource": "acs:oss:*:*:caiyuanrui-ap-southeast-1/test-backup/testbr1-2861990/*"
	    }
	  ]
	}
*/
func (s *serverImpl) testClassicBackupAccessAlicloud(ctx context.Context, input *TestClassicBackupAccessInput) (*TestClassicBackupAccessOutput, error) {
	u, err := url.Parse(input.BucketUri)
	if err != nil {
		tilog.Warn(ctx, "invalid bucket uri", zap.Error(err), zap.String("bucket_uri", utils.RedactURL(input.BucketUri)))
		return nil, status.Error(codes.InvalidArgument, "invalid bucket uri")
	}
	bucketName := u.Hostname()
	backupPrefix := u.Path

	cfg := oss.LoadDefaultConfig().
		WithCredentialsProvider(osscredentials.NewStaticCredentialsProvider(input.AccessKeyId, input.AccessKeySecret, "")).
		WithRegion(input.Region)
	client := oss.NewClient(cfg)

	// include bucket exsting test, get object test, backupmeta test
	backupMeta, resp, err := alicloudGetBackupMeta(ctx, bucketName, backupPrefix, client)
	if backupMeta == nil {
		tilog.Warn(ctx, "failed to get backup meta", zap.Error(err), zap.Any("resp", resp))
		return resp, err
	}

	// test list objects
	resp, err = alicloudTestListObjects(ctx, bucketName, backupPrefix, client)
	if err != nil || resp != nil {
		tilog.Warn(ctx, "failed to test list objects", zap.Error(err), zap.Any("resp", resp))
		return resp, err
	}

	resp = makeTestClassicBackupAccessOutput(backupMeta)
	resp.Region = input.Region
	tilog.Info(ctx, "test classic backup access succeeded", zap.Any("resp", resp))
	return resp, nil
}

func (s *serverImpl) testClassicBackupAccessAzure(ctx context.Context, input *TestClassicBackupAccessInput) (*TestClassicBackupAccessOutput, error) {
	u, err := url.Parse(input.BucketUri)
	if err != nil {
		tilog.Warn(ctx, "invalid bucket uri", zap.Error(err), zap.String("bucket_uri", utils.RedactURL(input.BucketUri)))
		return nil, status.Error(codes.InvalidArgument, "invalid bucket uri")
	}

	container, prefix, err := parseAzureBucketURI(u)
	if err != nil {
		tilog.Warn(ctx, "invalid azure bucket uri", zap.Error(err), zap.String("bucket_uri", utils.RedactURL(input.BucketUri)))
		return nil, status.Error(codes.InvalidArgument, "invalid bucket uri")
	}

	sasToken := strings.TrimPrefix(strings.TrimSpace(input.SasToken), "?")
	if sasToken == "" {
		return nil, status.Errorf(codes.InvalidArgument, "sas token is required")
	}

	accountURL := fmt.Sprintf("%s://%s", u.Scheme, u.Host)
	client, err := azblob.NewClientWithNoCredential(buildAzureServiceURL(accountURL, sasToken), nil)
	if err != nil {
		return nil, status.Error(codes.InvalidArgument, "invalid bucket uri")
	}

	resp, err := azureListBlobs(ctx, client, container, prefix)
	if err != nil || resp != nil {
		return resp, err
	}

	meta, resp, err := azureGetBackupMeta(ctx, client, container, path.Join(prefix, BackupMetaName))
	if meta == nil {
		return resp, err
	}

	return makeTestClassicBackupAccessOutput(meta), nil
}

func makeTestClassicBackupAccessOutput(meta *kvbackup.BackupMeta) *TestClassicBackupAccessOutput {
	out := &TestClassicBackupAccessOutput{
		TestClassicBackupAccessResponse: &apiv1.TestClassicBackupAccessResponse{
			Status: apiv1.TestClassicBackupAccessResponse_SUCCESS,
		},
	}
	out.Version = normailzeVersion(meta.ClusterVersion)
	out.BackupTime = timestamppb.New(time.UnixMilli(int64(meta.EndVersion >> 18)))
	out.DataSizeBytes = meta.BackupSize // single copy
	return out
}

/*
Normalize the version to "v7.5.6" format.
1. in 7.5.6, the original version is "\"v7.5.6\"\n"
2. in 8.5.4, the original version is "8.5.4"
*/
func normailzeVersion(version string) string {
	v := strings.TrimSpace(version)
	v = strings.Trim(v, "\"")
	v = strings.TrimPrefix(v, "v")
	return "v" + v
}

func normalizeBackupMetaPath(backupPrefix string) string {
	if backupPrefix == "/" {
		return BackupMetaName
	}
	return normalizePrefix(backupPrefix) + BackupMetaName
}

// remove lead slash and add trailing slash
func normalizePrefix(backupPrefix string) string {
	value := strings.TrimPrefix(backupPrefix, "/")
	value = strings.TrimSuffix(value, "/") + "/"
	return value
}

func awsBuildBucketClient(ctx context.Context, bucketName, ak, sk string) (*s3.Client, string, *TestClassicBackupAccessOutput, error) {
	cfg, err := config.LoadDefaultConfig(ctx,
		config.WithCredentialsProvider(credentials.NewStaticCredentialsProvider(ak, sk, "")),
		// Region is required but irrelevant; S3 will return the correct region
		config.WithRegion("us-east-1"),
	)
	if err != nil {
		return nil, "", nil, status.Errorf(codes.Internal, "failed to load s3 default config: %v", err)
	}
	client := s3.NewFromConfig(cfg)
	lresp, err := client.GetBucketLocation(ctx, &s3.GetBucketLocationInput{
		Bucket: aws.String(bucketName),
	})
	if err != nil {
		tilog.Warn(ctx, "failed to get bucket location", zap.Error(err), zap.String("bucket_name", bucketName))
		var ae smithy.APIError
		if errors.As(err, &ae) {
			switch ae.ErrorCode() {
			case NoSuchBucket:
				return nil, "", &TestClassicBackupAccessOutput{
					TestClassicBackupAccessResponse: &apiv1.TestClassicBackupAccessResponse{
						Status:  apiv1.TestClassicBackupAccessResponse_NO_SUCH_BUCKET,
						Message: "bucket not found",
					},
				}, nil
			case AccessDenied:
				return nil, "", &TestClassicBackupAccessOutput{
					TestClassicBackupAccessResponse: &apiv1.TestClassicBackupAccessResponse{
						Status:  apiv1.TestClassicBackupAccessResponse_ACCESS_DENIED,
						Message: "access denied when getting bucket location",
					},
				}, nil
			case InvalidAccessKeyId:
				return nil, "", &TestClassicBackupAccessOutput{
					TestClassicBackupAccessResponse: &apiv1.TestClassicBackupAccessResponse{
						Status:  apiv1.TestClassicBackupAccessResponse_ACCESS_DENIED,
						Message: "invalid access key id",
					},
				}, nil
			case SignatureDoesNotMatch:
				return nil, "", &TestClassicBackupAccessOutput{
					TestClassicBackupAccessResponse: &apiv1.TestClassicBackupAccessResponse{
						Status:  apiv1.TestClassicBackupAccessResponse_ACCESS_DENIED,
						Message: "invalid access key secret",
					},
				}, nil
			default:
				return nil, "", &TestClassicBackupAccessOutput{
					TestClassicBackupAccessResponse: &apiv1.TestClassicBackupAccessResponse{
						Status:  apiv1.TestClassicBackupAccessResponse_INVALID_PARAMS,
						Message: "failed to get bucket location: " + err.Error(),
					},
				}, nil
			}
		}
		return nil, "", nil, status.Errorf(codes.Internal, "unknown error, failed to get bucket location: %v", err)
	}
	region := string(lresp.LocationConstraint)
	// Notice: bucket region is empty if the bucket is in us-east-1, use default region "us-east-1"
	if region == "" {
		return client, "us-east-1", nil, nil
	}

	cfg, err = config.LoadDefaultConfig(ctx,
		config.WithCredentialsProvider(credentials.NewStaticCredentialsProvider(ak, sk, "")),
		config.WithRegion(region),
	)
	if err != nil {
		return nil, "", nil, status.Errorf(codes.Internal, "failed to load s3 default config: %v", err)
	}
	return s3.NewFromConfig(cfg), region, nil, nil
}

func parseAzureBucketURI(u *url.URL) (string, string, error) {
	if u.Host == "" || u.Scheme == "" {
		return "", "", fmt.Errorf("invalid bucket uri")
	}
	parts := strings.Split(strings.TrimPrefix(u.Path, "/"), "/")
	if len(parts) == 0 || parts[0] == "" {
		return "", "", fmt.Errorf("invalid bucket uri")
	}
	container := parts[0]
	prefix := strings.Join(parts[1:], "/")
	return container, prefix, nil
}

func buildAzureServiceURL(baseURL, sasToken string) string {
	u, err := url.Parse(baseURL)
	if err != nil {
		return baseURL
	}
	q := u.Query()
	if sasToken != "" {
		sasValues, err := url.ParseQuery(strings.TrimPrefix(sasToken, "?"))
		if err != nil {
			return baseURL
		}
		for k, vals := range sasValues {
			for _, v := range vals {
				q.Add(k, v)
			}
		}
	}
	u.RawQuery = q.Encode()
	return u.String()
}

func azureListBlobs(ctx context.Context, client *azblob.Client, container, prefix string) (*TestClassicBackupAccessOutput, error) {
	prefix = strings.TrimPrefix(prefix, "/")
	pager := client.NewListBlobsFlatPager(container, &azblob.ListBlobsFlatOptions{
		Prefix: &prefix,
	})
	if pager.More() {
		_, err := pager.NextPage(ctx)
		if err != nil {
			tilog.Warn(ctx, "failed to list blobs", zap.Error(err), zap.String("container", container), zap.String("prefix", prefix))
			return mapAzureListError(err)
		}
	}
	//nolint:nilnil // return clear resp if need user action, return error if it's unexpected error
	return nil, nil
}

func azureGetBackupMeta(ctx context.Context, client *azblob.Client, container, blobName string) (*kvbackup.BackupMeta, *TestClassicBackupAccessOutput, error) {
	blobName = strings.TrimPrefix(blobName, "/")
	resp, err := client.DownloadStream(ctx, container, blobName, nil)
	if err != nil {
		tilog.Warn(ctx, "failed to download stream", zap.Error(err), zap.String("container", container), zap.String("blob_name", blobName))
		out, err := mapAzureGetError(err)
		return nil, out, err
	}
	defer func() { _ = resp.Body.Close() }()

	body, err := io.ReadAll(resp.Body)
	if err != nil {
		return nil, nil, err
	}
	meta := new(kvbackup.BackupMeta)
	if err := proto.Unmarshal(body, meta); err != nil {
		return nil, &TestClassicBackupAccessOutput{
			TestClassicBackupAccessResponse: &apiv1.TestClassicBackupAccessResponse{
				Status:  apiv1.TestClassicBackupAccessResponse_ILLEGAL_BACKUP,
				Message: "invalid backup meta",
			},
		}, nil
	}
	return meta, nil, nil
}

func mapAzureListError(err error) (*TestClassicBackupAccessOutput, error) {
	var respErr *azcore.ResponseError
	if errors.As(err, &respErr) {
		switch respErr.StatusCode {
		case http.StatusNotFound:
			return &TestClassicBackupAccessOutput{
				TestClassicBackupAccessResponse: &apiv1.TestClassicBackupAccessResponse{
					Status:  apiv1.TestClassicBackupAccessResponse_NO_SUCH_BUCKET,
					Message: "no such bucket",
				},
			}, nil
		case http.StatusForbidden:
			return &TestClassicBackupAccessOutput{
				TestClassicBackupAccessResponse: &apiv1.TestClassicBackupAccessResponse{
					Status:  apiv1.TestClassicBackupAccessResponse_ACCESS_DENIED,
					Message: "access denied",
				},
			}, nil
		default:
			return &TestClassicBackupAccessOutput{
				TestClassicBackupAccessResponse: &apiv1.TestClassicBackupAccessResponse{
					Status:  apiv1.TestClassicBackupAccessResponse_INVALID_PARAMS,
					Message: respErr.Error(),
				},
			}, nil
		}
	}
	return nil, err
}

func mapAzureGetError(err error) (*TestClassicBackupAccessOutput, error) {
	var respErr *azcore.ResponseError
	if errors.As(err, &respErr) {
		switch respErr.StatusCode {
		case http.StatusNotFound:
			return &TestClassicBackupAccessOutput{
				TestClassicBackupAccessResponse: &apiv1.TestClassicBackupAccessResponse{
					Status:  apiv1.TestClassicBackupAccessResponse_ILLEGAL_BACKUP,
					Message: "backup meta not found",
				},
			}, nil
		case http.StatusForbidden:
			return &TestClassicBackupAccessOutput{
				TestClassicBackupAccessResponse: &apiv1.TestClassicBackupAccessResponse{
					Status:  apiv1.TestClassicBackupAccessResponse_ACCESS_DENIED,
					Message: "access denied",
				},
			}, nil
		default:
			return &TestClassicBackupAccessOutput{
				TestClassicBackupAccessResponse: &apiv1.TestClassicBackupAccessResponse{
					Status:  apiv1.TestClassicBackupAccessResponse_INVALID_PARAMS,
					Message: respErr.Error(),
				},
			}, nil
		}
	}
	return nil, err
}

func awsTestListObjects(ctx context.Context, bucketName, backupPrefix string, client *s3.Client) (*TestClassicBackupAccessOutput, error) {
	_, err := client.ListObjectsV2(ctx, &s3.ListObjectsV2Input{
		Bucket:  aws.String(bucketName),
		Prefix:  aws.String(normalizePrefix(backupPrefix)),
		MaxKeys: aws.Int32(1),
	})
	if err != nil {
		tilog.Warn(ctx, "failed to list objects", zap.Error(err), zap.String("bucket_name", bucketName), zap.String("backup_prefix", backupPrefix))
		var ae smithy.APIError
		if errors.As(err, &ae) {
			switch ae.ErrorCode() {
			case AccessDenied:
				return &TestClassicBackupAccessOutput{
					TestClassicBackupAccessResponse: &apiv1.TestClassicBackupAccessResponse{
						Status:  apiv1.TestClassicBackupAccessResponse_ACCESS_DENIED,
						Message: "access denied when listing objects",
					},
				}, nil
			default:
				return &TestClassicBackupAccessOutput{
					TestClassicBackupAccessResponse: &apiv1.TestClassicBackupAccessResponse{
						Status:  apiv1.TestClassicBackupAccessResponse_INVALID_PARAMS,
						Message: "failed to list objects: " + err.Error(),
					},
				}, nil
			}
		}
		return nil, status.Errorf(codes.Internal, "unknown error, failed to list objects: %v", err)
	}
	//nolint:nilnil // return clear resp if need user action, return error if it's unexpected error
	return nil, nil
}

func awsGetBackupMeta(ctx context.Context, bucketName, backupPrefix string, client *s3.Client) (*kvbackup.BackupMeta, *TestClassicBackupAccessOutput, error) {
	oresp, err := client.GetObject(ctx, &s3.GetObjectInput{
		Bucket: aws.String(bucketName),
		Key:    aws.String(normalizeBackupMetaPath(backupPrefix)),
	})
	if err != nil {
		tilog.Warn(ctx, "failed to get backup meta", zap.Error(err), zap.String("bucket_name", bucketName), zap.String("backup_prefix", backupPrefix))
		var ae smithy.APIError
		if errors.As(err, &ae) {
			switch ae.ErrorCode() {
			case NoSuchKey:
				return nil, &TestClassicBackupAccessOutput{
					TestClassicBackupAccessResponse: &apiv1.TestClassicBackupAccessResponse{
						Status:  apiv1.TestClassicBackupAccessResponse_ILLEGAL_BACKUP,
						Message: "backupmeta not found",
					},
				}, nil
			case AccessDenied:
				return nil, &TestClassicBackupAccessOutput{
					TestClassicBackupAccessResponse: &apiv1.TestClassicBackupAccessResponse{
						Status:  apiv1.TestClassicBackupAccessResponse_ACCESS_DENIED,
						Message: "access denied when getting backupmeta object",
					},
				}, nil

			default:
				return nil, &TestClassicBackupAccessOutput{
					TestClassicBackupAccessResponse: &apiv1.TestClassicBackupAccessResponse{
						Status:  apiv1.TestClassicBackupAccessResponse_INVALID_PARAMS,
						Message: "failed to get backupmeta: " + err.Error(),
					},
				}, nil
			}
		}
		return nil, nil, status.Errorf(codes.Internal, "unknown error, failed to get backupmeta: %v", err)
	}
	defer func() { _ = oresp.Body.Close() }()
	metaDataBytes, err := io.ReadAll(oresp.Body)
	if err != nil {
		return nil, nil, status.Errorf(codes.Internal, "failed to read backupmeta: %v", err)
	}
	backupMeta := &kvbackup.BackupMeta{}
	err = proto.Unmarshal(metaDataBytes, backupMeta)
	if err != nil {
		return nil, &TestClassicBackupAccessOutput{
			TestClassicBackupAccessResponse: &apiv1.TestClassicBackupAccessResponse{
				Status:  apiv1.TestClassicBackupAccessResponse_ILLEGAL_BACKUP,
				Message: "backupmeta is illegal, unmarshal failed: " + err.Error(),
			},
		}, nil
	}
	return backupMeta, nil, nil
}

func alicloudTestListObjects(ctx context.Context, bucketName, backupPrefix string, client *oss.Client) (*TestClassicBackupAccessOutput, error) {
	_, err := client.ListObjects(ctx, &oss.ListObjectsRequest{
		Bucket:  oss.Ptr(bucketName),
		Prefix:  oss.Ptr(normalizePrefix(backupPrefix)),
		MaxKeys: 1,
	})
	if err != nil {
		tilog.Warn(ctx, "failed to list objects", zap.Error(err), zap.String("bucket_name", bucketName), zap.String("backup_prefix", backupPrefix))
		var se *oss.ServiceError
		if errors.As(err, &se) {
			switch se.ErrorCode() {
			case AccessDenied:
				return &TestClassicBackupAccessOutput{
					TestClassicBackupAccessResponse: &apiv1.TestClassicBackupAccessResponse{
						Status:  apiv1.TestClassicBackupAccessResponse_ACCESS_DENIED,
						Message: "access denied when listing objects",
					},
				}, nil
			default:
				return &TestClassicBackupAccessOutput{
					TestClassicBackupAccessResponse: &apiv1.TestClassicBackupAccessResponse{
						Status:  apiv1.TestClassicBackupAccessResponse_INVALID_PARAMS,
						Message: "failed to list objects: " + err.Error(),
					},
				}, nil
			}
		}
		return nil, status.Errorf(codes.Internal, "unknown error, failed to list objects: %v", err)
	}
	//nolint:nilnil // return clear resp if need user action, return error if it's unexpected error
	return nil, nil
}

func alicloudGetBackupMeta(ctx context.Context, bucketName, backupPrefix string, client *oss.Client) (*kvbackup.BackupMeta, *TestClassicBackupAccessOutput, error) {
	resp, err := client.GetObject(ctx, &oss.GetObjectRequest{
		Bucket: oss.Ptr(bucketName),
		Key:    oss.Ptr(normalizeBackupMetaPath(backupPrefix)),
	})
	if err != nil {
		tilog.Warn(ctx, "failed to get backup meta", zap.Error(err), zap.String("bucket_name", bucketName), zap.String("backup_prefix", backupPrefix))
		var se *oss.ServiceError
		if errors.As(err, &se) {
			switch se.ErrorCode() {
			case NoSuchBucket:
				return nil, &TestClassicBackupAccessOutput{
					TestClassicBackupAccessResponse: &apiv1.TestClassicBackupAccessResponse{
						Status:  apiv1.TestClassicBackupAccessResponse_NO_SUCH_BUCKET,
						Message: "bucket not found",
					},
				}, nil
			case NoSuchKey:
				return nil, &TestClassicBackupAccessOutput{
					TestClassicBackupAccessResponse: &apiv1.TestClassicBackupAccessResponse{
						Status:  apiv1.TestClassicBackupAccessResponse_ILLEGAL_BACKUP,
						Message: "backupmeta not found",
					},
				}, nil
			case AccessDenied:
				return nil, &TestClassicBackupAccessOutput{
					TestClassicBackupAccessResponse: &apiv1.TestClassicBackupAccessResponse{
						Status:  apiv1.TestClassicBackupAccessResponse_ACCESS_DENIED,
						Message: "access denied when getting backupmeta object",
					},
				}, nil
			case InvalidAccessKeyId:
				return nil, &TestClassicBackupAccessOutput{
					TestClassicBackupAccessResponse: &apiv1.TestClassicBackupAccessResponse{
						Status:  apiv1.TestClassicBackupAccessResponse_ACCESS_DENIED,
						Message: "invalid access key id",
					},
				}, nil
			case SignatureDoesNotMatch:
				return nil, &TestClassicBackupAccessOutput{
					TestClassicBackupAccessResponse: &apiv1.TestClassicBackupAccessResponse{
						Status:  apiv1.TestClassicBackupAccessResponse_ACCESS_DENIED,
						Message: "invalid access key secret",
					},
				}, nil
			default:
				return nil, &TestClassicBackupAccessOutput{
					TestClassicBackupAccessResponse: &apiv1.TestClassicBackupAccessResponse{
						Status:  apiv1.TestClassicBackupAccessResponse_INVALID_PARAMS,
						Message: "failed to get backupmeta: " + err.Error(),
					},
				}, nil
			}
		}

		// regiona id of alicloud is part of the oss service url, so if the url is invalid, it will return "no such host" error
		if strings.Contains(err.Error(), "no such host") {
			return nil, &TestClassicBackupAccessOutput{
				TestClassicBackupAccessResponse: &apiv1.TestClassicBackupAccessResponse{
					Status:  apiv1.TestClassicBackupAccessResponse_INVALID_PARAMS,
					Message: "invalid region id",
				},
			}, nil
		}
		return nil, nil, status.Errorf(codes.Internal, "unknown error, failed to get backupmeta: %v", err)
	}
	defer func() { _ = resp.Body.Close() }()
	metaDataBytes, err := io.ReadAll(resp.Body)
	if err != nil {
		return nil, nil, status.Errorf(codes.Internal, "failed to read backupmeta: %v", err)
	}
	backupMeta := &kvbackup.BackupMeta{}
	err = proto.Unmarshal(metaDataBytes, backupMeta)
	if err != nil {
		return nil, &TestClassicBackupAccessOutput{
			TestClassicBackupAccessResponse: &apiv1.TestClassicBackupAccessResponse{
				Status:  apiv1.TestClassicBackupAccessResponse_ILLEGAL_BACKUP,
				Message: "backupmeta is illegal, unmarshal failed: " + err.Error(),
			},
		}, nil
	}
	return backupMeta, nil, nil
}