package gcp

import (
	"bytes"
	"context"
	"errors"
	"fmt"
	"io"
	"strings"
	"sync"

	"github.com/tidbcloud/observability/pkg/common/message"
	"google.golang.org/api/iterator"

	"cloud.google.com/go/iam"
	"cloud.google.com/go/storage"
	"github.com/tidbcloud/observability/pkg/common/metrics"
	"github.com/tidbcloud/observability/pkg/resourceworker/providers/gcp/gcs"
	"github.com/tidbcloud/observability/pkg/resourceworker/providers/lifecyclecompare"
	"github.com/tidbcloud/observability/pkg/resourceworker/providers/types"
)

const (
	cloudStorageClient    = "cloud-storage-client"
	ErrNotFound           = "no matching binding group found"
	errGetObjectFailed    = "get object failed"
	errPutObjectFailed    = "put object failed"
	errDeleteObjectFailed = "delete object failed"
	gcsMaxLifecycleRules  = 1000
)

var csonce sync.Once

// Clients should be reused instead of created as needed. The methods of Client are safe for concurrent use by multiple goroutines.
// Since the client will be reclaimed by the system at the end of the program's lifecycle, we do not have to call the cleanup process
var cscli *storage.Client

type CloudStorageClient struct {
	cli     gcs.CloudStorageClient
	project string
	region  string
}

func NewCloudStorage(project string, region string) *CloudStorageClient {
	csonce.Do(func() {
		cli, err := storage.NewClient(context.Background())
		if err != nil {
			//we will panic at first use if any error found
			panic(fmt.Sprintf("create cloud storage client failed in gcp vendor err: %s", err))
		}
		cscli = cli
	})

	return &CloudStorageClient{
		cli:     gcs.StorageClientWarp{Client: cscli},
		project: project,
		region:  region,
	}
}

func (cs *CloudStorageClient) CreateBucket(ctx context.Context, name string) error {
	attrs := &storage.BucketAttrs{
		Location: cs.region,
		UniformBucketLevelAccess: storage.UniformBucketLevelAccess{
			Enabled: true,
		},
	}
	bucket := cs.cli.Bucket(name)

	if err := bucket.Create(ctx, cs.project, attrs); err != nil {
		return fmt.Errorf("Bucket(%q).Create: %v", name, err)
	}
	return nil
}

func (cs *CloudStorageClient) BucketExists(ctx context.Context, name string) (bool, error) {
	bucket := cs.cli.Bucket(name)
	_, err := bucket.Attrs(ctx)
	if err != nil {
		if errors.Is(err, storage.ErrBucketNotExist) {
			return false, nil
		}
		return false, err
	}
	return true, nil
}

func (cs *CloudStorageClient) SetBucketPolicy(ctx context.Context, name string, policy *iam.Policy3) error {
	bucket := cs.cli.Bucket(name)
	if err := bucket.IAMV3().SetPolicy(ctx, policy); err != nil {
		return err
	}
	return nil
}

func (cs *CloudStorageClient) GetBucketPolicy(ctx context.Context, name string) (*iam.Policy3, bool, error) {
	bucket := cs.cli.Bucket(name)
	policy, err := bucket.IAMV3().Policy(ctx)
	if err != nil {
		if errors.Is(err, storage.ErrBucketNotExist) {
			return nil, false, nil
		}
		return nil, false, err
	}
	return policy, true, nil
}

func (cs *CloudStorageClient) DeleteBucketPolicy(ctx context.Context, name string, title string) error {
	bucket := cs.cli.Bucket(name)
	policy, err := bucket.IAMV3().Policy(ctx)
	if err != nil {
		return err
	}
	// Find the index of the binding matching inputs
	i := -1
	for j, binding := range policy.Bindings {
		if binding.Condition != nil {
			condition := binding.Condition
			if condition.Title == title {
				i = j
			}
		}
	}

	if i == -1 {
		return errors.New(ErrNotFound)
	}

	// Get a slice of the bindings, removing the binding at index i
	policy.Bindings = append(policy.Bindings[:i], policy.Bindings[i+1:]...)

	if err := bucket.IAMV3().SetPolicy(ctx, policy); err != nil {
		return err
	}
	return nil
}

func (cs *CloudStorageClient) Get(ctx context.Context, bucketName string, objectName string) (*io.ReadCloser, error) {
	reader, err := cs.cli.Bucket(bucketName).Object(objectName).NewReader(ctx)
	if err != nil {
		if errors.Is(err, storage.ErrObjectNotExist) {
			return nil, types.ObjectNotFoundError
		}
		metrics.HandleGCPAPIRequestMetrics(cs.region, cloudStorageClient, "GetObject", objectName, errGetObjectFailed)
		return nil, err
	}
	var r io.ReadCloser = reader
	return &r, nil
}

// ListObjects returns a list of objects in the bucket with the given prefix.
func (cs *CloudStorageClient) ListObjects(ctx context.Context, bucketName string, prefix string, options types.ListOptions) ([]string, error) {
	query := &storage.Query{Prefix: prefix}
	it := cs.cli.Bucket(bucketName).Objects(ctx, query)

	var objects []string
	count := int32(0)
	for {
		object, err := it.Next()
		if err == iterator.Done {
			break
		}
		if err != nil {
			return nil, fmt.Errorf("error when iterating objects at %q: %w", prefix, err)
		}
		objects = append(objects, object.Name)
		count++
		if options.MaxKeys != nil && count >= *options.MaxKeys {
			break
		}
	}
	return objects, nil
}

func (cs *CloudStorageClient) Put(ctx context.Context, bucketName string, objectName string, reader *bytes.Reader) error {
	writer := cs.cli.Bucket(bucketName).Object(objectName).NewWriter(ctx)
	defer writer.Close()
	_, err := reader.WriteTo(writer)
	if err != nil {
		metrics.HandleGCPAPIRequestMetrics(cs.region, cloudStorageClient, "PutObject", objectName, errPutObjectFailed)
		return err
	}
	// Close the Writer to complete the upload
	if err := writer.Close(); err != nil {
		return fmt.Errorf("Writer.Close: %w", err)
	}
	return nil
}

// buildGCSLifecycleRules helper function converts a LifecycleRuleParam into one or more GCS storage.LifecycleRule objects.
func buildGCSLifecycleRules(param message.LifecycleRuleParam) []storage.LifecycleRule {
	var rules []storage.LifecycleRule

	if param.ExpirationDays > 0 {
		rules = append(rules, storage.LifecycleRule{
			Action: storage.LifecycleAction{
				Type: storage.DeleteAction,
			},
			Condition: storage.LifecycleCondition{
				AgeInDays:     int64(param.ExpirationDays),
				MatchesPrefix: []string{param.Prefix},
			},
		})
	}

	if param.ColdDays > 0 {
		rules = append(rules, storage.LifecycleRule{
			Action: storage.LifecycleAction{
				Type:         storage.SetStorageClassAction,
				StorageClass: "COLDLINE",
			},
			Condition: storage.LifecycleCondition{
				AgeInDays:     int64(param.ColdDays),
				MatchesPrefix: []string{param.Prefix},
			},
		})
	}
	return rules
}

func gcsLifecycleRuleKey(rule storage.LifecycleRule) string {
	if len(rule.Condition.MatchesPrefix) == 0 {
		return ""
	}
	key := fmt.Sprintf("%s_%s", rule.Condition.MatchesPrefix[0], rule.Action.Type)
	if rule.Action.Type == storage.SetStorageClassAction {
		key = fmt.Sprintf("%s_%s_%s", rule.Condition.MatchesPrefix[0], rule.Action.Type, rule.Action.StorageClass)
	}
	return key
}

func isManagedClusterLifecyclePrefix(prefix string) bool {
	return strings.HasSuffix(prefix, "/profiles/") ||
		strings.HasSuffix(prefix, "/oom_record/") ||
		strings.HasSuffix(prefix, "/regions/")
}

func isPrefixInNewGCSRules(prefix string, newRulesMap map[string]storage.LifecycleRule) bool {
	for _, rule := range newRulesMap {
		if len(rule.Condition.MatchesPrefix) > 0 && rule.Condition.MatchesPrefix[0] == prefix {
			return true
		}
	}
	return false
}

func shouldDropExistingGCSRule(oldRule storage.LifecycleRule, newRulesMap map[string]storage.LifecycleRule, unusedRulePrefixes []string) bool {
	if len(oldRule.Condition.MatchesPrefix) == 0 {
		return true
	}
	prefix := oldRule.Condition.MatchesPrefix[0]

	if _, exists := newRulesMap[gcsLifecycleRuleKey(oldRule)]; exists {
		return true
	}

	for _, unusedPrefix := range unusedRulePrefixes {
		if strings.HasPrefix(prefix, unusedPrefix) {
			return true
		}
	}

	// GCS lifecycle rules have no stable IDs, so drop stale per-cluster rules by object path pattern.
	if isManagedClusterLifecyclePrefix(prefix) && !isPrefixInNewGCSRules(prefix, newRulesMap) {
		return true
	}

	return false
}

func collectNewGCSRules(rules []message.LifecycleRuleParam) (map[string]storage.LifecycleRule, []string) {
	newRulesMap := make(map[string]storage.LifecycleRule)
	newRuleOrder := make([]string, 0, len(rules)*2)
	for _, param := range rules {
		gcsRules := buildGCSLifecycleRules(param)
		for _, gcsRule := range gcsRules {
			key := gcsLifecycleRuleKey(gcsRule)
			if _, exists := newRulesMap[key]; !exists {
				newRuleOrder = append(newRuleOrder, key)
			}
			newRulesMap[key] = gcsRule
		}
	}
	return newRulesMap, newRuleOrder
}

func (cs *CloudStorageClient) AddLifecycle(ctx context.Context, bucketName string, rules []message.LifecycleRuleParam, unusedRulePrefixes []string) error {
	bkt := cs.cli.Bucket(bucketName)

	existingRules, err := bkt.Attrs(ctx)
	if err != nil {
		return fmt.Errorf("failed to get existing lifecycle rules for bucket %s: %w", bucketName, err)
	}

	newRulesMap, newRuleOrder := collectNewGCSRules(rules)
	rulesToApply := make([]storage.LifecycleRule, 0, len(newRulesMap)+len(existingRules.Lifecycle.Rules))
	for _, key := range newRuleOrder {
		rulesToApply = append(rulesToApply, newRulesMap[key])
	}

	for _, oldRule := range existingRules.Lifecycle.Rules {
		if shouldDropExistingGCSRule(oldRule, newRulesMap, unusedRulePrefixes) {
			continue
		}
		rulesToApply = append(rulesToApply, oldRule)
	}

	existingKept := len(rulesToApply) - len(newRulesMap)
	if len(rulesToApply) > gcsMaxLifecycleRules {
		fmt.Printf(
			"gcs lifecycle: truncating rules for bucket %s from %d to %d (new=%d, existing_kept=%d)\n",
			bucketName, len(rulesToApply), gcsMaxLifecycleRules, len(newRulesMap), existingKept,
		)
		rulesToApply = rulesToApply[:gcsMaxLifecycleRules]
	} else {
		fmt.Printf(
			"gcs lifecycle: applying %d rules to bucket %s (new=%d, existing_kept=%d)\n",
			len(rulesToApply), bucketName, len(newRulesMap), existingKept,
		)
	}

	if lifecyclecompare.GCSLifecycleRulesEqual(rulesToApply, existingRules.Lifecycle.Rules) {
		fmt.Printf("lifecycle rules unchanged for bucket %s, skip update\n", bucketName)
		return nil
	}

	_, err = bkt.Update(ctx, storage.BucketAttrsToUpdate{
		Lifecycle: &storage.Lifecycle{
			Rules: rulesToApply,
		},
	})
	if err != nil {
		return fmt.Errorf("failed to set lifecycle rules for bucket %s: %w", bucketName, err)
	}

	return nil
}

func (cs *CloudStorageClient) RemoveLifecycle(ctx context.Context, bucketName string, name string) error {
	panic("implement me")
}

// Delete removes objects from the bucket.
func (cs *CloudStorageClient) Delete(ctx context.Context, bucketName string, objectNames ...string) error {
	for _, objectName := range objectNames {
		err := cs.cli.Bucket(bucketName).Object(objectName).Delete(ctx)
		if err != nil {
			metrics.HandleGCPAPIRequestMetrics(cs.region, cloudStorageClient, "DeleteObject", objectName, errDeleteObjectFailed)
			return err
		}
	}
	return nil
}