package storage

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

	"cloud.google.com/go/storage"
	"google.golang.org/api/impersonate"
	"google.golang.org/api/iterator"
	"google.golang.org/api/option"

	"github.com/tidbcloud/service-proxy/pkg/checker/core"
)

func init() {
	core.DefaultRegistry.Register("gcs", func(req *core.CheckRequest) (core.Checker, error) {
		if req.GCS == nil {
			return nil, errors.New("gcs params required")
		}
		return NewGCSChecker(req.GCS), nil
	})
}

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

// GCSChecker checks Google Cloud Storage connectivity.
type GCSChecker struct {
	params *core.GCSParams
}

// NewGCSChecker creates a new GCS checker.
func NewGCSChecker(params *core.GCSParams) core.Checker {
	return &GCSChecker{params: params}
}

// Type returns the checker type.
func (c *GCSChecker) Type() string {
	return "gcs"
}

// Check performs GCS connectivity check.
func (c *GCSChecker) Check(ctx context.Context) error {
	u, client, err := c.newClientWithCredentials(ctx)
	if err != nil {
		return err
	}

	// test storage.buckets.get
	attrs, err := client.Bucket(u.Bucket).Attrs(ctx)
	if err != nil {
		return fmt.Errorf("failed to get bucket attrs: %w", err)
	}

	// 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 errors.Is(err, iterator.Done) {
			break
		}
	}
	return checkBucketRegion(c.params.Region, attrs)
}

func checkBucketRegion(region string, attrs *storage.BucketAttrs) error {
	if region == "" {
		return nil
	}

	bucketRegions := extractBucketRegions(attrs)

	if isRegionCompatible(region, bucketRegions) {
		return nil
	}

	return fmt.Errorf("cluster region is not compatible with gcs bucket region")
}

// extractBucketRegions extracts region list from bucket attributes.
func extractBucketRegions(attrs *storage.BucketAttrs) []string {
	// Dual-region bucket
	if attrs.CustomPlacementConfig != nil {
		regions := make([]string, 0, len(attrs.CustomPlacementConfig.DataLocations))
		for _, r := range attrs.CustomPlacementConfig.DataLocations {
			regions = append(regions, strings.ToLower(r))
		}
		return regions
	}
	// Single-region or multi-region bucket
	return []string{strings.ToLower(attrs.Location)}
}

// isRegionCompatible checks if the cluster region is compatible with bucket regions.
func isRegionCompatible(region string, bucketRegions []string) bool {
	// Multi-region bucket (e.g., "us", "eu") - use prefix matching
	if isMultiRegionBucket(bucketRegions) {
		prefix := bucketRegions[0]
		return region == prefix || strings.HasPrefix(region, prefix+"-")
	}
	// Single-region or dual-region bucket - exact match
	for _, r := range bucketRegions {
		if r == region {
			return true
		}
	}
	return false
}

// isMultiRegionBucket checks if the bucket is a multi-region bucket.
func isMultiRegionBucket(regions []string) bool {
	return len(regions) == 1 && !strings.Contains(regions[0], "-")
}

func newGcsURL(gcsURL string) (*bucketURL, error) {
	gcsURL = strings.TrimSpace(gcsURL)
	u, err := url.Parse(gcsURL)
	if err != nil {
		return nil, fmt.Errorf("failed to parse gcs url: %w", 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 &bucketURL{
		URL:    gcsURL,
		Prefix: prefix,
		Bucket: bucket,
	}, nil
}

func (c *GCSChecker) newClientWithCredentials(ctx context.Context) (*bucketURL, *storage.Client, error) {
	u, err := newGcsURL(c.params.Url)
	if err != nil {
		return nil, nil, err
	}

	var 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 c.params.ImpersonateServiceAccount != "" {
		tokenSource, err := impersonate.CredentialsTokenSource(ctx, impersonate.CredentialsConfig{
			TargetPrincipal: c.params.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", c.params.ImpersonateServiceAccount, err)
		}
		opts = append(opts, option.WithTokenSource(tokenSource))
	}

	client, err := storage.NewClient(ctx, opts...)
	if err != nil {
		return nil, nil, fmt.Errorf("failed to create gcs client: %w", err)
	}
	return u, client, err
}