package gcs

import (
	"context"
	"fmt"
	"io"
	"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 (
	// this file is write by infra-provider
	CredentialsFilePath = "/root/gcp/key.json"
)

type GcsURL struct {
	URL    string
	Prefix string
	Bucket string
}

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" {
		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, impersonateServiceAccount string) (*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 len(impersonateServiceAccount) != 0 {
		tokenSource, err := impersonate.CredentialsTokenSource(ctx, impersonate.CredentialsConfig{
			TargetPrincipal: 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", 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
}

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

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

	object := ""
	// test storage.objects.list
	listQuery := client.Bucket(u.Bucket).Objects(ctx, &storage.Query{Prefix: u.Prefix})
	for {
		attrs, err := listQuery.Next()
		// test one objet is enough
		if err == iterator.Done || object != "" {
			break
		}
		if err != nil {
			return errors.AddStack(err)
		}
		object = attrs.Name
	}
	if object == "" {
		return errors.Errorf("no files found under the given GCS location (bucket=%q, folder=%q)", u.Bucket, u.Prefix)
	}
	// test storage.objects.get
	if _, err := client.Bucket(u.Bucket).Object(object).Attrs(ctx); err != nil {
		return errors.AddStack(err)
	}
	return nil
}

func NewGCSClient(ctx context.Context) (*storage.Client, error) {
	client, err := storage.NewClient(ctx)
	if err != nil {
		return nil, errors.AddStack(err)
	}
	return client, nil
}

func GetFileContent(ctx context.Context, gcsUrl, impersonateServiceAccount string, start, end int64) (string, error) {
	gcsURL, client, err := NewClientWithCredentials(ctx, gcsUrl, impersonateServiceAccount)
	if err != nil {
		return "", err
	}

	bucket := client.Bucket(gcsURL.Bucket)
	length := end - start + 1
	prefix := strings.TrimSuffix(gcsURL.Prefix, "/")
	reader, err := bucket.Object(prefix).NewRangeReader(ctx, start, length)
	if err != nil {
		return "", errors.Annotate(err, "new range reader")
	}

	ret := make([]byte, length)
	if _, err = reader.Read(ret); err != nil && errors.Cause(err) != io.EOF {
		return "", errors.Annotate(err, "failed to read object")
	}
	return string(ret), nil
}

// GetBucketRegions returns the regions of the bucket in lower case.
// single region bucket: return []string{us-west1}
// multi region bucket: return []string{us}
// dual region bucket: return []string{us-east1, us-west1}
func GetBucketRegions(ctx context.Context, gcsURL, impersonateServiceAccount string) ([]string, error) {
	u, client, err := NewClientWithCredentials(ctx, gcsURL, impersonateServiceAccount)
	if err != nil {
		return nil, err
	}
	attrs, err := client.Bucket(u.Bucket).Attrs(ctx)
	if err != nil {
		return nil, err
	}

	// for dual region bucket
	if attrs.CustomPlacementConfig != nil {
		regions := make([]string, 0, len(attrs.CustomPlacementConfig.DataLocations))
		for _, region := range attrs.CustomPlacementConfig.DataLocations {
			regions = append(regions, strings.ToLower(region))
		}
		return regions, nil
	}
	return []string{strings.ToLower(attrs.Location)}, nil
}