package alicloud

import (
	"bytes"
	"context"
	"errors"
	"fmt"
	"io"
	"net/http"
	"strings"

	"github.com/tidbcloud/observability/pkg/common/message"
	"github.com/tidbcloud/observability/pkg/resourceworker/providers/lifecyclecompare"
	"github.com/tidbcloud/observability/pkg/resourceworker/providers/types"

	"github.com/aliyun/alibabacloud-oss-go-sdk-v2/oss"
	"github.com/tidbcloud/observability/pkg/common/metrics"
)

const (
	s3ClientLabel            = "s3-client"
	errCreateBucketFailed    = "create bucket failed"
	errHeadBucketFailed      = "determine if a bucket exists failed"
	errGetObjectFailed       = "get object failed"
	errPutObjectFailed       = "put object failed"
	errGetLifecycleFailed    = "get lifecycle failed"
	errPutLifecycleFailed    = "put lifecycle failed"
	errDeleteLifecycleFailed = "delete lifecycle failed"
)

type S3Client struct {
	cli            OssClient
	cfg            oss.Config
	redundancyType oss.DataRedundancyType
}

func NewS3(cfg *oss.Config, redundancyType string) S3Client {
	c := oss.NewClient(cfg)
	//if not set , use ZRS as the default redundancy type
	if redundancyType == "" {
		redundancyType = string(oss.DataRedundancyZRS)
	}
	return S3Client{
		cli:            c,
		cfg:            *cfg,
		redundancyType: oss.DataRedundancyType(redundancyType),
	}
}

// CreateBucket used to create a new bucket in the S3
func (s3 S3Client) CreateBucket(ctx context.Context, name string) error {
	input := &oss.PutBucketRequest{
		Bucket: oss.Ptr(name),
		Acl:    oss.BucketACLPrivate,
		CreateBucketConfiguration: &oss.CreateBucketConfiguration{
			DataRedundancyType: s3.redundancyType,
		},
	}
	_, err := s3.cli.PutBucket(ctx, input)
	if err != nil {
		metrics.HandleAWSAPIRequestMetrics(*s3.cfg.Region, s3ClientLabel, "CreateBucket", name, errCreateBucketFailed)
	}
	return err
}

// BucketExists returns if a bucket exists in the S3
func (s3 S3Client) BucketExists(ctx context.Context, name string) (bool, error) {
	ok, err := s3.cli.IsBucketExist(ctx, name)
	if err != nil {
		metrics.HandleAWSAPIRequestMetrics(*s3.cfg.Region, s3ClientLabel, "HeadBucket", name, errHeadBucketFailed)
		return false, err
	}
	if !ok {
		return false, nil

	}
	return true, nil
}

func (s3 S3Client) Get(ctx context.Context, bucketName string, objectName string) (*io.ReadCloser, error) {
	out, err := s3.cli.GetObject(ctx, &oss.GetObjectRequest{
		Bucket: oss.Ptr(bucketName),
		Key:    oss.Ptr(objectName),
	})
	if err != nil {
		var serviceError *oss.ServiceError
		if errors.As(err, &serviceError) && (serviceError.Code == "NoSuchKey" || serviceError.StatusCode == http.StatusNotFound) {
			return nil, types.ObjectNotFoundError
		}
		metrics.HandleAWSAPIRequestMetrics(*s3.cfg.Region, s3ClientLabel, "GetObject", objectName, errGetObjectFailed)
		return nil, err
	}
	return &out.Body, nil
}

func (s3 S3Client) Put(ctx context.Context, bucketName string, objectName string, reader *bytes.Reader) error {
	_, err := s3.cli.PutObject(ctx, &oss.PutObjectRequest{
		Bucket: oss.Ptr(bucketName),
		Key:    oss.Ptr(objectName),
		Body:   reader,
	})
	if err != nil {
		metrics.HandleAWSAPIRequestMetrics(*s3.cfg.Region, s3ClientLabel, "PutObject", objectName, errPutObjectFailed)
		return err
	}
	return nil
}

func (s3 S3Client) Delete(ctx context.Context, bucketName string, objectNames ...string) error {
	objects := make([]oss.DeleteObject, len(objectNames))
	for i, objectName := range objectNames {
		objects[i] = oss.DeleteObject{
			Key: oss.Ptr(objectName),
		}
	}
	_, err := s3.cli.DeleteMultipleObjects(ctx, &oss.DeleteMultipleObjectsRequest{
		Bucket:  oss.Ptr(bucketName),
		Objects: objects,
	})
	if err != nil {
		return err
	}
	return nil
}

func (s3 S3Client) ListObjects(ctx context.Context, bucketName string, prefix string, options types.ListOptions) ([]string, error) {
	listReq := &oss.ListObjectsV2Request{
		Bucket: oss.Ptr(bucketName),
		Prefix: oss.Ptr(prefix),
	}
	if options.MaxKeys != nil {
		listReq.MaxKeys = *options.MaxKeys
	}
	paginator := s3.cli.NewListObjectsV2Paginator(listReq)
	var objects []string
	for paginator.HasNext() {
		page, err := paginator.NextPage(ctx)
		if err != nil {
			return nil, err
		}
		for _, object := range page.Contents {
			objects = append(objects, *object.Key)
		}
	}
	return objects, nil
}

func MockS3(cli OssClient) S3Client {
	return S3Client{
		cli: cli,
	}
}

func (s3 S3Client) AddLifecycle(ctx context.Context, bucketName string, rules []message.LifecycleRuleParam, unusedRulePrefixes []string) error {
	// Get existing lifecycle configuration
	getReq := &oss.GetBucketLifecycleRequest{
		Bucket: oss.Ptr(bucketName),
	}
	getRes, err := s3.cli.GetBucketLifecycle(ctx, getReq)
	if err != nil {
		// If the error is not "lifecycle configuration not found", return the error
		var serviceError *oss.ServiceError
		if errors.As(err, &serviceError) && (serviceError.Code == "NoSuchLifecycleConfiguration" || serviceError.StatusCode == http.StatusNotFound) {
			// Lifecycle configuration does not exist, use empty rules list
			getRes = &oss.GetBucketLifecycleResult{
				LifecycleConfiguration: &oss.LifecycleConfiguration{
					Rules: []oss.LifecycleRule{},
				},
			}
		} else {
			metrics.HandleAWSAPIRequestMetrics(*s3.cfg.Region, s3ClientLabel, "GetLifecycle", bucketName, errGetLifecycleFailed)
			return err
		}
	}

	// Build new rules
	newRules := make([]oss.LifecycleRule, 0)
	for _, rule := range rules {
		newRules = append(newRules, buildOSSLifecycleRule(rule.RuleName, rule.Prefix, rule.ColdDays, rule.ExpirationDays))
	}

	// Merge existing rules (exclude rules to be deleted and rules with the same name as new rules)
	existingRules := []oss.LifecycleRule{}
	if getRes.LifecycleConfiguration != nil && getRes.LifecycleConfiguration.Rules != nil {
		existingRules = getRes.LifecycleConfiguration.Rules
	}

	newRulePrefixes := make([]string, 0, len(newRules))
	for _, rule := range newRules {
		if p := ossLifecycleRulePrefix(rule); p != "" {
			newRulePrefixes = append(newRulePrefixes, p)
		}
	}

	for _, oldRule := range existingRules {
		drop := false
		// Check if it matches unusedRulePrefixes
		if oldRule.ID != nil {
			for _, prefix := range unusedRulePrefixes {
				if strings.HasPrefix(*oldRule.ID, prefix) {
					drop = true
					break
				}
			}
		}
		// Check if it has the same name as new rules
		if !drop && oldRule.ID != nil {
			for _, newRule := range newRules {
				if newRule.ID != nil && *newRule.ID == *oldRule.ID {
					drop = true
					break
				}
			}
		}
		// OSS rejects duplicate or overlapping prefixes across rules.
		if !drop {
			oldPrefix := ossLifecycleRulePrefix(oldRule)
			for _, newPrefix := range newRulePrefixes {
				if lifecyclePrefixesConflict(oldPrefix, newPrefix) {
					drop = true
					break
				}
			}
		}
		if !drop {
			newRules = append(newRules, oldRule)
		}
	}

	newRules = dedupeOSSLifecycleRulesByPrefix(newRules)

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

	// Update lifecycle configuration
	putReq := &oss.PutBucketLifecycleRequest{
		Bucket:                 oss.Ptr(bucketName),
		AllowSameActionOverlap: oss.Ptr("true"),
		LifecycleConfiguration: &oss.LifecycleConfiguration{
			Rules: newRules,
		},
	}
	_, err = s3.cli.PutBucketLifecycle(ctx, putReq)
	if err != nil {
		metrics.HandleAWSAPIRequestMetrics(*s3.cfg.Region, s3ClientLabel, "PutLifecycle", bucketName, errPutLifecycleFailed)
		return err
	}

	return nil
}

func ossLifecycleRulePrefix(rule oss.LifecycleRule) string {
	if rule.Prefix != nil && *rule.Prefix != "" {
		return *rule.Prefix
	}
	return ""
}

func lifecyclePrefixesConflict(a, b string) bool {
	if a == "" || b == "" {
		return false
	}
	if a == b {
		return true
	}
	return strings.HasPrefix(a, b) || strings.HasPrefix(b, a)
}

func dedupeOSSLifecycleRulesByPrefix(rules []oss.LifecycleRule) []oss.LifecycleRule {
	out := make([]oss.LifecycleRule, 0, len(rules))
	seenPrefixes := make(map[string]struct{}, len(rules))
	for _, rule := range rules {
		prefix := ossLifecycleRulePrefix(rule)
		if prefix == "" {
			out = append(out, rule)
			continue
		}
		if _, exists := seenPrefixes[prefix]; exists {
			continue
		}
		seenPrefixes[prefix] = struct{}{}
		out = append(out, rule)
	}
	return out
}

func buildOSSLifecycleRule(ruleName string, prefix string, coldDays int32, expirationDays int32) oss.LifecycleRule {
	status := "Enabled"
	rule := oss.LifecycleRule{
		ID:     oss.Ptr(ruleName),
		Prefix: oss.Ptr(prefix),
		Status: &status,
	}

	if expirationDays > 0 {
		rule.Expiration = &oss.LifecycleRuleExpiration{
			Days: oss.Ptr(expirationDays),
		}
	}

	if coldDays > 0 {
		// OSS cold storage types: IA (Infrequent Access), Archive, ColdArchive
		// Use IA (Infrequent Access) as the cold storage type here
		storageClass := oss.StorageClassType("IA")
		rule.Transitions = []oss.LifecycleRuleTransition{
			{
				Days:         oss.Ptr(coldDays),
				StorageClass: storageClass,
			},
		}
	}

	return rule
}

func (s3 S3Client) RemoveLifecycle(ctx context.Context, bucketName string, name string) error {
	// Get existing lifecycle configuration
	getReq := &oss.GetBucketLifecycleRequest{
		Bucket: oss.Ptr(bucketName),
	}
	getRes, err := s3.cli.GetBucketLifecycle(ctx, getReq)
	if err != nil {
		var serviceError *oss.ServiceError
		if errors.As(err, &serviceError) && (serviceError.Code == "NoSuchLifecycleConfiguration" || serviceError.StatusCode == http.StatusNotFound) {
			// Lifecycle configuration does not exist, return success directly
			return nil
		}
		metrics.HandleAWSAPIRequestMetrics(*s3.cfg.Region, s3ClientLabel, "GetLifecycle", bucketName, errGetLifecycleFailed)
		return err
	}

	// Filter out rules to be deleted
	if getRes.LifecycleConfiguration == nil || getRes.LifecycleConfiguration.Rules == nil {
		return nil
	}

	filteredRules := make([]oss.LifecycleRule, 0)
	for _, rule := range getRes.LifecycleConfiguration.Rules {
		if rule.ID == nil || *rule.ID != name {
			filteredRules = append(filteredRules, rule)
		}
	}

	// If there are no rules left, delete the entire lifecycle configuration
	if len(filteredRules) == 0 {
		deleteReq := &oss.DeleteBucketLifecycleRequest{
			Bucket: oss.Ptr(bucketName),
		}
		_, err = s3.cli.DeleteBucketLifecycle(ctx, deleteReq)
		if err != nil {
			metrics.HandleAWSAPIRequestMetrics(*s3.cfg.Region, s3ClientLabel, "DeleteLifecycle", bucketName, errDeleteLifecycleFailed)
			return err
		}
		return nil
	}

	// Update lifecycle configuration, remove the specified rule
	putReq := &oss.PutBucketLifecycleRequest{
		Bucket:                 oss.Ptr(bucketName),
		AllowSameActionOverlap: oss.Ptr("true"),
		LifecycleConfiguration: &oss.LifecycleConfiguration{
			Rules: filteredRules,
		},
	}
	_, err = s3.cli.PutBucketLifecycle(ctx, putReq)
	if err != nil {
		metrics.HandleAWSAPIRequestMetrics(*s3.cfg.Region, s3ClientLabel, "PutLifecycle", bucketName, errPutLifecycleFailed)
		return err
	}

	return nil
}