package central

import (
	"context"
	"fmt"
	"net/url"
	"os"
	"strings"

	"cloud.google.com/go/storage"
	"github.com/pingcap/errors"
	"google.golang.org/api/impersonate"
	"google.golang.org/api/iterator"
	"google.golang.org/api/option"
)

const (
	credentialsFilePath = "/root/gcp/key.json"
)

type GcsURL struct {
	URL    string
	Prefix string
	Bucket string
}

type GcsOptions struct {
	ImpersonateServiceAccount *string
}

// see https://docs.pingcap.com/tidbcloud/migrate-from-amazon-s3-or-gcs#step-2-configure-gcs-access
func checkGcsAccess(ctx context.Context, gcsURL string, region string, gcsOpt *GcsOptions) error {
	u, client, err := newClientWithCredentials(ctx, gcsURL, gcsOpt)
	if err != nil {
		return err
	}

	// test storage.buckets.get
	attrs, err := client.Bucket(u.Bucket).Attrs(ctx)
	if err != nil {
		return errors.AddStack(err)
	}

	object := ""
	// test storage.objects.list
	listQuery := client.Bucket(u.Bucket).Objects(ctx, &storage.Query{Prefix: u.Prefix})
	for {
		_, err := listQuery.Next()
		// test one objet is enough
		if err == iterator.Done || object != "" {
			break
		}
	}
	return checkBucketRegion(region, attrs)
}

func checkBucketRegion(region string, attrs *storage.BucketAttrs) error {
	var bucketRegions []string
	if region != "" {
		// for dual region bucket
		if attrs.CustomPlacementConfig != nil {
			bucketRegions = make([]string, 0, len(attrs.CustomPlacementConfig.DataLocations))
			for _, region := range attrs.CustomPlacementConfig.DataLocations {
				bucketRegions = append(bucketRegions, strings.ToLower(region))
			}
		} else {
			bucketRegions = []string{strings.ToLower(attrs.Location)}
		}
		// for multi-region bucket, regions will be like ["us"]
		if len(bucketRegions) == 1 && !strings.Contains(bucketRegions[0], "-") {
			if strings.Contains(region, bucketRegions[0]) {
				return nil
			}
		}
		// for dual-region bucket, regions will be like ["us-east1", "us-west1"]
		// for single-region bucket, regions will be like ["us-west1"]
		for _, r := range bucketRegions {
			if r == region {
				return nil
			}
		}
		return fmt.Errorf("cluster region %s not same as gcs bucket region %+v", region, bucketRegions)
	}
	return nil
}

func newGcsURL(gcsURL string) (*GcsURL, error) {
	gcsURL = strings.TrimSpace(gcsURL)
	u, err := url.Parse(gcsURL)
	if err != nil {
		return nil, errors.AddStack(err)
	}

	if u.Scheme != "gs" && u.Scheme != "gcs" {
		return nil, errors.New("an invalid gcs endpoint URL")
	}
	bucket := u.Host
	prefix := u.Path
	prefix = strings.TrimLeft(prefix, "/")
	if prefix != "" && !strings.HasSuffix(prefix, "/") {
		prefix += "/"
	}
	return &GcsURL{
		URL:    gcsURL,
		Prefix: prefix,
		Bucket: bucket,
	}, nil
}

func newClientWithCredentials(
	ctx context.Context, gcsURL string, gcsOpt *GcsOptions,
) (*GcsURL, *storage.Client, error) {
	u, err := newGcsURL(gcsURL)
	if err != nil {
		return nil, nil, err
	}

	opts := []option.ClientOption{}
	if _, err := os.Stat(credentialsFilePath); err == nil {
		// use this opt only when file `CredentialsFilePath` exists
		// for some runtime gke, can get creds from node sa and don't relay on this file.
		opts = append(opts, option.WithCredentialsFile(credentialsFilePath))
	}
	if gcsOpt.ImpersonateServiceAccount != nil {
		tokenSource, err := impersonate.CredentialsTokenSource(ctx, impersonate.CredentialsConfig{
			TargetPrincipal: *gcsOpt.ImpersonateServiceAccount,
			Scopes:          []string{"https://www.googleapis.com/auth/devstorage.read_only"},
		}, opts...)
		if err != nil {
			return nil, nil, fmt.Errorf(
				"impersonate to iam service account %s error: %w", *gcsOpt.ImpersonateServiceAccount, err)
		}
		opts = append(opts, option.WithTokenSource(tokenSource))
	}

	client, err := storage.NewClient(ctx, opts...)
	if err != nil {
		return nil, nil, errors.AddStack(err)
	}
	return u, client, err
}