package cloud_storage

import (
	"context"
	"encoding/base64"
	"errors"
	"fmt"
	"net/url"
	"strings"

	"cloud.google.com/go/storage"
	"github.com/tidbcloud/pkgv2/bizerr"
	"github.com/tidbcloud/pkgv2/log"
	"go.uber.org/zap"
	"google.golang.org/api/iterator"
	"google.golang.org/api/option"
)

type GCS struct {
	Client *storage.Client
	Bucket string
	Prefix *string
}

type GCSConfig struct {
	URI        string
	AccountKey *string
	client     *storage.Client
}

type GCSOption func(*GCSConfig)

func WithGCSAccountKey(accountKey string) GCSOption {
	return func(cfg *GCSConfig) {
		cfg.AccountKey = &accountKey
	}
}

func WithGcsClient(client *storage.Client) GCSOption {
	return func(cfg *GCSConfig) {
		cfg.client = client
	}
}

func NewGCS(ctx context.Context, uri string, opts ...GCSOption) (CloudStorage, bizerr.BizError) {
	cfg := &GCSConfig{URI: uri}
	for _, opt := range opts {
		opt(cfg)
	}

	bucket, prefix, bizErr := parseGCSParams(cfg.URI)
	if bizErr != nil {
		return nil, bizErr
	}
	if cfg.client != nil {
		// If a client is provided, use it directly
		return &GCS{
			Bucket: bucket,
			Client: cfg.client,
			Prefix: prefix,
		}, nil
	}

	if cfg.AccountKey == nil {
		return nil, bizerr.New(bizerr.ErrCodeParams, "GCSConfig: must provide accountKey")
	}
	d, err := DecodeBase64(*cfg.AccountKey)
	if err != nil {
		return nil, bizerr.New(bizerr.ErrCodeParams, "invalid service account key")
	}

	client, err := storage.NewClient(ctx, option.WithCredentialsJSON(d))
	if err != nil {
		log.Error(ctx, "failed to create client", zap.Error(err))
		return nil, bizerr.New(ErrCodeGcsAccessDeny, errMsgGcsAccessDeny)
	}

	return &GCS{
		Bucket: bucket,
		Client: client,
		Prefix: prefix,
	}, nil
}

func (g *GCS) ListAll(ctx context.Context) ([]string, bizerr.BizError) {
	var objectPaths []string
	bucket := g.Client.Bucket(g.Bucket)
	query := &storage.Query{}
	if g.Prefix != nil {
		query.Prefix = *g.Prefix
	}

	it := bucket.Objects(ctx, query)
	for {
		attrs, err := it.Next()
		if errors.Is(err, iterator.Done) {
			break
		}
		if err != nil {
			log.Error(ctx, "failed to list objects", zap.Error(err))
			if errors.Is(err, context.DeadlineExceeded) {
				return nil, bizerr.New(bizerr.ErrCodeTimeout, "")
			}
			return nil, bizerr.New(ErrCodeGcsAccessDeny, errMsgGcsAccessDeny)
		}
		objectPaths = append(objectPaths, fmt.Sprintf("gs://%s/%s", g.Bucket, attrs.Name))
	}
	return objectPaths, nil
}

func (g *GCS) CheckImportPermission(ctx context.Context) bizerr.BizError {
	bucket := g.Client.Bucket(g.Bucket)
	query := &storage.Query{}
	if g.Prefix != nil {
		query.Prefix = *g.Prefix
	}
	// check list access
	it := bucket.Objects(ctx, query)
	obj, err := it.Next()
	if err != nil {
		if errors.Is(err, iterator.Done) {
			if isOneFile(g.Prefix) {
				return bizerr.New(ErrCodeImportDataFileNotFound, "Can't find data file: %s", *g.Prefix)
			}
			return bizerr.New(ErrCodeImportEmptyFolder, errMsgImportEmptyFolder)
		}
		log.Error(ctx, "failed to list objects", zap.Error(err))
		return bizerr.New(ErrCodeGcsAccessDeny, errMsgGcsAccessDeny)
	}

	var name string
	if isOneFile(g.Prefix) {
		name = *g.Prefix
	} else {
		name = obj.Name
	}
	// check get access
	if _, err := g.Client.Bucket(g.Bucket).Object(name).Attrs(ctx); err != nil {
		if errors.Is(err, storage.ErrObjectNotExist) {
			return bizerr.New(ErrCodeImportDataFileNotFound, "Can't find data file: %s", name)
		}
		log.Error(ctx, "failed to get object", zap.Error(err))
		return bizerr.New(ErrCodeGcsAccessDeny, errMsgGcsAccessDeny)
	}

	return nil
}

func (s *GCS) RetrieveSchemaFile(ctx context.Context, schema string, table string, subDir string) (*string, bizerr.BizError) {
	bucket := s.Client.Bucket(s.Bucket)
	schemaFilePath := getSchemaFilePath(s.Prefix, subDir, schema, table)

	// Query objects with the base prefix
	query := &storage.Query{
		Prefix: schemaFilePath,
	}

	it := bucket.Objects(ctx, query)
	for {
		attrs, err := it.Next()
		if errors.Is(err, iterator.Done) {
			break
		}
		if err != nil {
			log.Error(ctx, "failed to list objects from GCS", zap.Error(err))
			if errors.Is(err, context.DeadlineExceeded) {
				return nil, bizerr.New(bizerr.ErrCodeTimeout, "")
			}
			return nil, bizerr.New(bizerr.ErrCodeUnknown, err.Error())
		}

		if attrs.Name != "" && matchSchemaFileExtension(attrs.Name, schemaFilePath) {
			result := getRelativeFilePath(s.Prefix, attrs.Name)
			return &result, nil
		}
	}

	return nil, nil // No matching files found
}

func (s *GCS) ListLimited(ctx context.Context, opts *ListFilesOptions) ([]string, bool, bizerr.BizError) {
	maxResults := GetMaxFilesFromOptions(opts)

	bucket := s.Client.Bucket(s.Bucket)
	query := &storage.Query{}

	if s.Prefix != nil {
		query.Prefix = *s.Prefix
	}

	it := bucket.Objects(ctx, query)
	pager := iterator.NewPager(it, maxResults, "")
	var allFiles []*storage.ObjectAttrs

	nextPageToken, err := pager.NextPage(&allFiles)
	if err != nil {
		log.Error(ctx, "failed to list objects from GCS", zap.Error(err))
		if errors.Is(err, context.DeadlineExceeded) {
			return nil, false, bizerr.New(bizerr.ErrCodeTimeout, "")
		}
		return nil, false, bizerr.New(bizerr.ErrCodeUnknown, err.Error())
	}

	var files []string
	for _, attrs := range allFiles {
		if !strings.HasSuffix(attrs.Name, "/") {
			files = append(files, getRelativeFilePath(s.Prefix, attrs.Name))
		}
	}

	truncated := nextPageToken != ""
	return files, truncated, nil
}

func parseGCSParams(uri string) (bucket string, prefix *string, _ bizerr.BizError) {
	p, err := url.Parse(EscapePath(uri))
	if err != nil {
		return "", nil, bizerr.New(bizerr.ErrCodeParams, "Invalid URL")
	}
	if p.Host == "" {
		return "", nil, bizerr.New(bizerr.ErrCodeParams, "Invalid URL: Bucket name is missing or improperly specified.")
	}
	pre := strings.TrimPrefix(p.Path, "/")
	if pre != "" {
		return p.Host, &pre, nil
	}
	return p.Host, nil, nil
}

func isOneFile(prefix *string) bool {
	if prefix == nil {
		return false
	}
	return !strings.HasSuffix(*prefix, "/")
}

// DecodeBase64 decodes a Base64 URL encoded string.
// If the input string's length is a multiple of 4, it uses the standard Base64 decoding
// with padding. Otherwise, it uses the raw Base64 decoding without padding.
func DecodeBase64(blob string) ([]byte, error) {
	if len(blob)%4 == 0 {
		// decode with padding
		return base64.StdEncoding.DecodeString(blob)
	}
	// decode without padding
	return base64.RawStdEncoding.DecodeString(blob)
}