package impl

import (
	"context"
	"fmt"
	"strings"

	"github.com/AlekSi/pointer"
	"github.com/aws/aws-sdk-go/aws"
	"github.com/aws/aws-sdk-go/aws/awserr"
	"github.com/aws/aws-sdk-go/service/s3"
	"github.com/elliotchance/pie/v2"
	"github.com/tidbcloud/dataflow-service/app/models/bucket"
	"github.com/tidbcloud/dataflow-service/app/models/domain"
	dfs3 "github.com/tidbcloud/dataflow-service/pkg/aws"
	"github.com/tidbcloud/dataflow-service/pkg/log"
	"github.com/tidbcloud/pkgv2/bizerr"
	"go.uber.org/zap"
)

var _ bucket.AWSBucketManagerModel = (*awsBucketManagerImpl)(nil)

const defaultBucketReplicationPriority = 10

type awsBucketManagerImpl struct {
	clientProxy dfs3.AWSBucketClientProxy
}

func NewAWSBucketManagerModel(
	clientProxy dfs3.AWSBucketClientProxy,
) bucket.AWSBucketManagerModel {
	return &awsBucketManagerImpl{
		clientProxy: clientProxy,
	}
}

func (b *awsBucketManagerImpl) CheckBucketIfExists(ctx context.Context, req *domain.AWSBucketReq) bizerr.BizError {
	c, err := b.clientProxy.GetClient(ctx, req.ToGetAWSClientReq())
	if err != nil {
		return bizerr.WrapError(err, "get client by aws account: %s meet error", req.AWSAccountID)
	}
	_, getErr := c.HeadBucket(&s3.HeadBucketInput{Bucket: pointer.ToString(req.Name)})
	if getErr != nil {
		return bizerr.WrapError(getErr, "get bucket")
	}
	return nil
}

func (b *awsBucketManagerImpl) CreateBucket(ctx context.Context, req *domain.CreateAWSBucketReq) bizerr.BizError {
	ctx = log.WithTraceID(ctx)
	c, err := b.clientProxy.GetClient(ctx, req.ToGetAWSClientReq())
	if err != nil {
		return bizerr.WrapError(err, "get client by aws account: %s meet error", req.AWSAccountID)
	}

	//  create bucket with region location constraint
	createReq := &s3.CreateBucketInput{
		Bucket:          pointer.ToString(req.Name),
		ObjectOwnership: pointer.ToString("BucketOwnerPreferred"),
	}
	// https://docs.aws.amazon.com/zh_cn/AmazonS3/latest/API/API_control_CreateBucket.html
	// us-east-1 is invalid, if we want to create in us-east-1, we should not set LocationConstraint
	if req.Region != "us-east-1" {
		createReq.CreateBucketConfiguration = &s3.CreateBucketConfiguration{
			LocationConstraint: pointer.ToString(req.Region),
		}
	}

	_, createErr := c.CreateBucket(createReq)
	log.GetLogger(ctx).Info("create bucket", zap.Any("req", createReq))
	if createErr != nil {
		log.GetLogger(ctx).Warn("create bucket", zap.Error(createErr))
		return bizerr.WrapError(createErr, "create bucket")
	}
	return nil
}

func (b *awsBucketManagerImpl) EnableBucketEncryption(
	ctx context.Context, req *domain.EnableAWSBucketEncryptionReq,
) bizerr.BizError {
	ctx = log.WithTraceID(ctx)
	c, err := b.clientProxy.GetClient(ctx, req.ToGetAWSClientReq())
	if err != nil {
		return bizerr.WrapError(err, "get client by aws account: %s meet error", req.AWSAccountID)
	}

	// Set Default Encryption
	rule := &s3.ServerSideEncryptionByDefault{
		SSEAlgorithm: pointer.ToString("AES256"),
	}
	if req.AwsKmsArn != nil {
		rule.SSEAlgorithm = pointer.ToString("aws:kms")
		rule.KMSMasterKeyID = req.AwsKmsArn
	}
	putReq := &s3.PutBucketEncryptionInput{
		Bucket: pointer.ToString(req.Name),
		ServerSideEncryptionConfiguration: &s3.ServerSideEncryptionConfiguration{
			Rules: []*s3.ServerSideEncryptionRule{{ApplyServerSideEncryptionByDefault: rule}},
		},
	}
	_, encErr := c.PutBucketEncryption(putReq)
	log.GetLogger(ctx).Info("put bucket encryption", zap.Any("req", putReq))
	if encErr != nil {
		log.GetLogger(ctx).Warn("put bucket encryption", zap.Any("err", encErr))
		return bizerr.WrapError(encErr, "put bucket encryption")
	}
	return nil
}

func (b *awsBucketManagerImpl) EnableBucketVersioning(
	ctx context.Context, req *domain.EnableAWSBucketVersioningReq,
) bizerr.BizError {
	ctx = log.WithTraceID(ctx)
	c, err := b.clientProxy.GetClient(ctx, req.ToGetAWSClientReq())
	if err != nil {
		return bizerr.WrapError(err, "get client by aws account: %s meet error", req.AWSAccountID)
	}
	// Enable versioning
	putReq := &s3.PutBucketVersioningInput{
		Bucket: pointer.ToString(req.Name),
		VersioningConfiguration: &s3.VersioningConfiguration{
			Status: pointer.ToString("Enabled"),
		},
	}
	log.GetLogger(ctx).Info("put bucket versioning", zap.Any("req", putReq))
	_, versionErr := c.PutBucketVersioning(putReq)
	if versionErr != nil {
		log.GetLogger(ctx).Warn("put bucket versioning", zap.Any("err", versionErr))
		return bizerr.WrapError(versionErr, "put bucket versioning")
	}
	return nil
}

func (b *awsBucketManagerImpl) PutBucketTags(
	ctx context.Context, req *domain.PutAWSBucketTagsReq,
) bizerr.BizError {
	ctx = log.WithTraceID(ctx)
	c, err := b.clientProxy.GetClient(ctx, req.ToGetAWSClientReq())
	if err != nil {
		return bizerr.WrapError(err, "get client by aws account: %s meet error", req.AWSAccountID)
	}

	var tagSet []*s3.Tag
	for k, v := range req.Tags {
		tagSet = append(tagSet, &s3.Tag{Key: pointer.ToString(k), Value: pointer.ToString(v)})
	}

	putReq := &s3.PutBucketTaggingInput{
		Bucket:  pointer.ToString(req.Name),
		Tagging: &s3.Tagging{TagSet: tagSet},
	}
	// Set Bucket Tagging
	_, tagErr := c.PutBucketTagging(putReq)
	log.GetLogger(ctx).Info("put bucket tagging", zap.Any("req", putReq))
	if tagErr != nil {
		log.GetLogger(ctx).Warn("put bucket tagging", zap.Any("err", err))
		return bizerr.WrapError(tagErr, "put bucket tagging")
	}
	return nil
}

func (b *awsBucketManagerImpl) AddAllowRoleArnToBucketPolicy(
	ctx context.Context, req *domain.AddAWSAllowRoleArnToBucketPolicyReq,
) (*domain.AddAWSAllowRoleArnToBucketPolicyResp, bizerr.BizError) {
	ctx = log.WithTraceID(ctx)
	c, err := b.clientProxy.GetClient(ctx, req.ToGetAWSClientReq())
	if err != nil {
		return nil, bizerr.WrapError(err, "get client by aws account: %s meet error", req.AWSAccountID)
	}

	getResp, err := c.GetBucketPolicy(&s3.GetBucketPolicyInput{
		Bucket: pointer.ToString(req.Name),
	})
	if err != nil && !strings.Contains(err.Error(), "NoSuchBucketPolicy") {
		return nil, bizerr.WrapError(err, "get bucket policy")
	}

	var policy *string
	if getResp != nil {
		policy = getResp.Policy
	}
	updateBucketPolicy, iamARNs, err := AddRolesToBucketPolicy(ctx, req.Name, policy, req.IAMRoleARNs)
	if err != nil {
		return nil, bizerr.WrapError(err, "build bucket policy")
	}

	putReq := &s3.PutBucketPolicyInput{
		Bucket: pointer.ToString(req.Name),
		Policy: updateBucketPolicy,
	}
	log.GetLogger(ctx).Info("put bucket policy", zap.Any("req", putReq))
	_, err = c.PutBucketPolicy(putReq)
	if err != nil {
		log.GetLogger(ctx).Warn("put bucket policy meet error", zap.Any("err", err))
		return nil, bizerr.WrapError(err, "put bucket policy")
	}
	return &domain.AddAWSAllowRoleArnToBucketPolicyResp{
		IAMRoleARNs: iamARNs,
	}, nil
}

func (b *awsBucketManagerImpl) DeleteRoleArnFromBucketPolicy(
	ctx context.Context, req *domain.DeleteAWSRoleArnFromBucketPolicyReq,
) (*domain.DeleteAWSRoleArnFromBucketPolicyResp, bizerr.BizError) {
	ctx = log.WithTraceID(ctx)
	c, err := b.clientProxy.GetClient(ctx, req.ToGetAWSClientReq())
	if err != nil {
		return nil, bizerr.WrapError(err, "get client by aws account: %s meet error", req.AWSAccountID)
	}

	getResp, err := c.GetBucketPolicy(&s3.GetBucketPolicyInput{
		Bucket: pointer.ToString(req.Name),
	})
	if err != nil && !strings.Contains(err.Error(), "NoSuchBucketPolicy") {
		return nil, bizerr.WrapError(err, "get bucket policy")
	}

	var policy *string
	if getResp != nil {
		policy = getResp.Policy
	}
	_, iamARNs, err := AddRolesToBucketPolicy(ctx, req.Name, policy, nil)
	if err != nil {
		return nil, bizerr.WrapError(err, "build bucket policy")
	}
	if len(iamARNs) == 0 {
		return &domain.DeleteAWSRoleArnFromBucketPolicyResp{}, nil
	}

	updateBucketPolicy, iamARNs, err := ModifyRolesToBucketPolicy(ctx, req.Name, policy, iamARNs, req.IAMRoleARNs)
	if err != nil {
		return nil, bizerr.WrapError(err, "build bucket policy")
	}
	putReq := &s3.PutBucketPolicyInput{
		Bucket: pointer.ToString(req.Name),
		Policy: updateBucketPolicy,
	}
	log.GetLogger(ctx).Info("put bucket policy", zap.Any("req", putReq))
	_, err = c.PutBucketPolicy(putReq)
	if err != nil {
		log.GetLogger(ctx).Warn("put bucket policy meet error", zap.Any("err", err))
		return nil, bizerr.WrapError(err, "put bucket policy")
	}
	return &domain.DeleteAWSRoleArnFromBucketPolicyResp{
		IAMRoleARNs: iamARNs,
	}, nil
}

func isBucketNotFoundErr(err error) bool {
	awsErr, ok := err.(awserr.Error)
	if !ok {
		return false
	}
	return awsErr.Code() == s3.ErrCodeNoSuchBucket
}

func (b *awsBucketManagerImpl) DeleteBucketIgnoreNotFound(
	ctx context.Context, req *domain.AWSBucketReq,
) bizerr.BizError {
	ctx = log.WithTraceID(ctx)
	c, err := b.clientProxy.GetClient(ctx, req.ToGetAWSClientReq())
	if err != nil {
		return bizerr.WrapError(err, "get client by aws account: %s meet error", req.AWSAccountID)
	}

	_, deleteErr := c.DeleteBucket(&s3.DeleteBucketInput{
		Bucket: pointer.ToString(req.Name),
	})

	if deleteErr != nil && !isBucketNotFoundErr(deleteErr) {
		log.GetLogger(ctx).Warn("delete bucket", zap.Any("err", deleteErr))
		return bizerr.WrapError(deleteErr, "delete bucket")
	}
	return nil
}

func (b *awsBucketManagerImpl) AddBucketLifecycleConfiguration(
	ctx context.Context, req *domain.AddBucketLifecycleConfigurationReq,
) bizerr.BizError {
	ctx = log.WithTraceID(ctx)
	c, err := b.clientProxy.GetClient(ctx, req.ToGetAWSClientReq())
	if err != nil {
		return bizerr.WrapError(err, "get client by aws account: %s meet error", req.AWSAccountID)
	}

	configuration, err := c.GetBucketLifecycleConfiguration(&s3.GetBucketLifecycleConfigurationInput{
		Bucket: pointer.ToOrNil(req.Name),
	})
	if err != nil && !strings.Contains(err.Error(), "NoSuchLifecycleConfiguration") {
		return bizerr.WrapError(err, "get bucket life cycleConfiguration policy")
	}

	var lifecycleRules []*s3.LifecycleRule
	if configuration != nil && len(configuration.Rules) > 0 {
		lifecycleRules = configuration.Rules
	}

	lifecycleRules = pie.Of(lifecycleRules).
		Filter(func(rule *s3.LifecycleRule) bool {
			if rule.ID == nil {
				return true
			}
			return *req.LifecycleRule.ID != *rule.ID
		}).
		Result

	lifecycleRules = append(lifecycleRules, &s3.LifecycleRule{
		ID: req.LifecycleRule.ID,
		NoncurrentVersionExpiration: &s3.NoncurrentVersionExpiration{
			NewerNoncurrentVersions: req.LifecycleRule.NoncurrentVersionExpiration.NewerNoncurrentVersions,
			NoncurrentDays:          req.LifecycleRule.NoncurrentVersionExpiration.NoncurrentDays,
		},
		Expiration: &s3.LifecycleExpiration{
			ExpiredObjectDeleteMarker: pointer.To(true),
		},
		AbortIncompleteMultipartUpload: &s3.AbortIncompleteMultipartUpload{
			DaysAfterInitiation: pointer.To(int64(1)),
		},
		Filter: &s3.LifecycleRuleFilter{},
		Status: aws.String("Enabled"),
	})

	putReq := &s3.PutBucketLifecycleConfigurationInput{
		Bucket: pointer.ToOrNil(req.Name),
		LifecycleConfiguration: &s3.BucketLifecycleConfiguration{
			Rules: lifecycleRules,
		},
	}
	log.GetLogger(ctx).Info("put bucket life cycleConfiguration", zap.Any("req", putReq))
	_, err = c.PutBucketLifecycleConfiguration(putReq)
	if err != nil {
		log.GetLogger(ctx).Warn("put bucket life cycleConfiguration meet error", zap.Any("err", err))
		return bizerr.WrapError(err, "put bucket life cycleConfiguration")
	}
	return nil
}

func (b *awsBucketManagerImpl) AddBucketReplication(
	ctx context.Context, req *domain.AddAWSBucketReplicationReq,
) (*domain.AddAWSBucketReplicationResp, bizerr.BizError) {
	ctx = log.WithTraceID(ctx)
	c, err := b.clientProxy.GetClient(ctx, req.ToGetAWSClientReq())
	if err != nil {
		return nil, bizerr.WrapError(err, "get client by aws account: %s meet error", req.AWSAccountID)
	}

	queryResp, err := c.GetBucketReplication(&s3.GetBucketReplicationInput{
		Bucket: pointer.ToOrNil(req.Name),
	})
	if err != nil && !strings.Contains(err.Error(), "ReplicationConfigurationNotFoundError") {
		return nil, bizerr.WrapError(err, "get bucket replication policy")
	}

	replicationRules := b.buildReplicationRules(req, queryResp)
	putReq := &s3.PutBucketReplicationInput{
		Bucket: pointer.ToOrNil(req.Name),
		ReplicationConfiguration: &s3.ReplicationConfiguration{
			Role:  pointer.ToString(domain.BuildDataflowServiceInfraRoleARN(req.AWSAccountID)),
			Rules: replicationRules,
		},
	}
	log.GetLogger(ctx).Info("put bucket replication policy", zap.Any("req", putReq))
	_, err = c.PutBucketReplication(putReq)
	if err != nil {
		log.GetLogger(ctx).Warn("put bucket replication policy meet error", zap.Any("err", err))
		return nil, bizerr.WrapError(err, "put bucket replication policy")
	}
	ruleIDs := pie.Of(replicationRules).
		Filter(func(rule *s3.ReplicationRule) bool {
			return rule.ID != nil
		}).
		StringsUsing(func(rule *s3.ReplicationRule) string {
			return *rule.ID
		})
	return &domain.AddAWSBucketReplicationResp{
		RuleIDs: ruleIDs,
	}, nil
}

func (b *awsBucketManagerImpl) buildReplicationRules(
	req *domain.AddAWSBucketReplicationReq, queryResp *s3.GetBucketReplicationOutput,
) []*s3.ReplicationRule {
	var replicationRules []*s3.ReplicationRule
	if queryResp != nil && queryResp.ReplicationConfiguration != nil &&
		len(queryResp.ReplicationConfiguration.Rules) > 0 {
		replicationRules = queryResp.ReplicationConfiguration.Rules
	}

	var maxPriority int64 = defaultBucketReplicationPriority
	pie.Of(replicationRules).Each(func(rule *s3.ReplicationRule) {
		if rule.Priority != nil && *rule.Priority > maxPriority {
			maxPriority = *rule.Priority
		}
	})
	replicationRules = pie.Of(replicationRules).
		Filter(func(rule *s3.ReplicationRule) bool {
			if rule.ID == nil {
				return true
			}
			return *req.ReplicationRule.ID != *rule.ID
		}).
		Result
	newRule := &s3.ReplicationRule{
		ID: req.ReplicationRule.ID,
		DeleteMarkerReplication: &s3.DeleteMarkerReplication{
			Status: aws.String("Enabled"),
		},
		Destination: &s3.Destination{
			Bucket: pointer.To(fmt.Sprintf("arn:aws:s3:::%s", *req.ReplicationRule.Destination.Bucket)),
		},
		Filter: &s3.ReplicationRuleFilter{
			Prefix: req.ReplicationRule.Filter.Prefix,
		},
		Status: aws.String("Enabled"),
	}
	// https://docs.aws.amazon.com/zh_cn/AmazonS3/latest/userguide/replication-walkthrough1.html
	// very important: The prefix must end with a forward slash (/). otherwise, the replication rule will not work.
	if !strings.HasSuffix(*newRule.Filter.Prefix, "/") {
		newRule.Filter.Prefix = pointer.To(fmt.Sprintf("%s/", *newRule.Filter.Prefix))
	}
	newRule.Priority = pointer.To(maxPriority + 1)
	replicationRules = append(replicationRules, newRule)
	return replicationRules
}

func (b *awsBucketManagerImpl) DeleteBucketReplication(
	ctx context.Context, req *domain.DeleteAWSBucketReplicationReq,
) (*domain.DeleteAWSBucketReplicationResp, bizerr.BizError) {
	ctx = log.WithTraceID(ctx)
	c, err := b.clientProxy.GetClient(ctx, req.ToGetAWSClientReq())
	if err != nil {
		return nil, bizerr.WrapError(err, "get client by aws account: %s meet error", req.AWSAccountID)
	}

	queryResp, err := c.GetBucketReplication(&s3.GetBucketReplicationInput{
		Bucket: pointer.ToOrNil(req.Name),
	})
	if err != nil {
		if !strings.Contains(err.Error(), "ReplicationConfigurationNotFoundError") {
			return nil, bizerr.WrapError(err, "get bucket replication policy")
		}
		return &domain.DeleteAWSBucketReplicationResp{}, nil
	}

	replicationRules := b.buildDeleteReplicationRules(req, queryResp)
	if len(replicationRules) == 0 {
		_, err = c.DeleteBucketReplication(&s3.DeleteBucketReplicationInput{
			Bucket: pointer.ToOrNil(req.Name),
		})
		if err != nil {
			log.GetLogger(ctx).Warn("delete bucket replication policy meet error", zap.Any("err", err))
			return nil, bizerr.WrapError(err, "delete bucket replication policy")
		}
		log.GetLogger(ctx).Warn("delete replication policy finished")
		return &domain.DeleteAWSBucketReplicationResp{}, nil
	}
	putReq := &s3.PutBucketReplicationInput{
		Bucket: pointer.ToOrNil(req.Name),
		ReplicationConfiguration: &s3.ReplicationConfiguration{
			Role:  pointer.ToString(domain.BuildDataflowServiceInfraRoleARN(req.AWSAccountID)),
			Rules: replicationRules,
		},
	}
	log.GetLogger(ctx).Info("put bucket replication policy", zap.Any("req", putReq))
	_, err = c.PutBucketReplication(putReq)
	if err != nil {
		log.GetLogger(ctx).Warn("put bucket replication policy meet error", zap.Any("err", err))
		return nil, bizerr.WrapError(err, "put bucket replication policy")
	}
	ruleIDs := pie.Of(replicationRules).
		Filter(func(rule *s3.ReplicationRule) bool {
			return rule.ID != nil
		}).
		StringsUsing(func(rule *s3.ReplicationRule) string {
			return *rule.ID
		})
	return &domain.DeleteAWSBucketReplicationResp{
		RuleIDs: ruleIDs,
	}, nil
}

func (b *awsBucketManagerImpl) buildDeleteReplicationRules(
	req *domain.DeleteAWSBucketReplicationReq, queryResp *s3.GetBucketReplicationOutput,
) []*s3.ReplicationRule {
	if queryResp == nil ||
		queryResp.ReplicationConfiguration == nil ||
		len(queryResp.ReplicationConfiguration.Rules) == 0 {
		return nil
	}

	return pie.Of(queryResp.ReplicationConfiguration.Rules).
		Filter(func(rule *s3.ReplicationRule) bool {
			if rule.ID == nil {
				return true
			}
			return *req.RuleID != *rule.ID
		}).
		Result
}

func (b *awsBucketManagerImpl) ModifyReplicationRTC(
	ctx context.Context, req *domain.ModifyReplicationRTCReq,
) bizerr.BizError {
	const (
		S3RTCMinutes = 15 // it's a constant required by AWS, refer to https://docs.aws.amazon.com/AmazonS3/latest/userguide/replication-time-control.html
	)
	ctx = log.WithTraceID(ctx)
	c, err := b.clientProxy.GetClient(ctx, req.ToGetAWSClientReq())
	if err != nil {
		return bizerr.WrapError(err, "get client by aws account: %s meet error", req.AWSAccountID)
	}

	queryResp, err := c.GetBucketReplication(&s3.GetBucketReplicationInput{
		Bucket: pointer.ToOrNil(req.Name),
	})
	if err != nil {
		if strings.Contains(err.Error(), "ReplicationConfigurationNotFoundError") {
			log.GetLogger(ctx).Warn("no replication configuration found, skipping RTC modification",
				zap.String("bucket", req.Name), zap.String("ruleID", req.RuleID))
			return nil
		}
		return bizerr.WrapError(err, "get bucket replication policy")
	}

	if queryResp == nil || queryResp.ReplicationConfiguration == nil ||
		len(queryResp.ReplicationConfiguration.Rules) == 0 {
		log.GetLogger(ctx).Warn("empty replication configuration, skipping RTC modification",
			zap.String("bucket", req.Name), zap.String("ruleID", req.RuleID))
		return bizerr.New(bizerr.ErrCodeUnknown, "empty replication configuration")
	}

	var found bool
	for _, rule := range queryResp.ReplicationConfiguration.Rules {
		if rule.ID == nil || *rule.ID != req.RuleID {
			continue
		}
		found = true
		if req.EnableRTC {
			rule.Destination.ReplicationTime = &s3.ReplicationTime{
				Status: aws.String("Enabled"),
				Time:   &s3.ReplicationTimeValue{Minutes: aws.Int64(S3RTCMinutes)},
			}
			rule.Destination.Metrics = &s3.Metrics{
				Status:         aws.String("Enabled"),
				EventThreshold: &s3.ReplicationTimeValue{Minutes: aws.Int64(S3RTCMinutes)},
			}
		} else {
			rule.Destination.ReplicationTime = nil
			rule.Destination.Metrics = nil
		}
		break
	}
	if !found {
		log.GetLogger(ctx).Warn("replication rule not found, skipping RTC modification",
			zap.String("bucket", req.Name), zap.String("ruleID", req.RuleID))
		return nil
	}

	putReq := &s3.PutBucketReplicationInput{
		Bucket:                   pointer.ToOrNil(req.Name),
		ReplicationConfiguration: queryResp.ReplicationConfiguration,
	}
	log.GetLogger(ctx).Info("put bucket replication policy (RTC modification)", zap.Any("req", putReq))
	_, err = c.PutBucketReplication(putReq)
	if err != nil {
		log.GetLogger(ctx).Warn("put bucket replication policy meet error", zap.Any("err", err))
		return bizerr.WrapError(err, "put bucket replication policy")
	}
	return nil
}