package eks

import (
	"context"
	"fmt"
	"time"

	"github.com/aws/aws-sdk-go-v2/service/s3/types"
	"github.com/aws/aws-sdk-go/aws"
	s3common "github.com/crossplane-contrib/provider-aws/apis/s3/common"
	s3v1alpha3 "github.com/crossplane-contrib/provider-aws/apis/s3/v1alpha3"
	s3v1beta1 "github.com/crossplane-contrib/provider-aws/apis/s3/v1beta1"
	xpv1 "github.com/crossplane/crossplane-runtime/apis/common/v1"
	"github.com/crossplane/crossplane-runtime/pkg/meta"
	corev1 "k8s.io/api/core/v1"
	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
	"k8s.io/utils/pointer"
	"sigs.k8s.io/controller-runtime/pkg/client"

	eksv1alpha1 "github.com/tidbcloud/eks-provider/api/v1alpha1"
	"github.com/tidbcloud/eks-provider/controllers/tiflow/task/cluster"
	"github.com/tidbcloud/eks-provider/controllers/tiflow/task/cluster/master"
	"github.com/tidbcloud/infra-provider/pkg/deploy"
	"github.com/tidbcloud/infra-provider/pkg/k8s"
	"github.com/tidbcloud/infra-provider/pkg/namer"
	"github.com/tidbcloud/infra-provider/pkg/procedure"
	"github.com/tidbcloud/infra-provider/pkg/util"
)

type Bucket struct {
	*master.BaseTask
}

// NewBucket create Bucket instance.
func NewBucket(ctx *master.Context) *Bucket {
	return &Bucket{BaseTask: &master.BaseTask{Context: ctx}}
}

var _ deploy.Task = (*Bucket)(nil)

// Name implements deploy.Task.Name interface.
func (b *Bucket) Name() string {
	return "Bucket"
}

// Create implements deploy.Task interface.
func (b *Bucket) Create(ctx context.Context) deploy.Result {
	b.BeginAt = util.Pointer(time.Now())
	name := cluster.FormatBucketName(b.Config.FullManage.Project, b.Network, b.Namer.GetTiflowClusterName())
	tags := cluster.TagsForBucket(b.Network, name)

	spec, err := eksv1alpha1.EKSNetworkSpecForNetwork(b.Network)
	if err != nil {
		return deploy.Error(err)
	}
	if spec.NodeRoleArn == "" {
		return deploy.Error(fmt.Errorf("node role arn is empty"))
	}

	cluster.AddBucketAnnotationToMaster(b.TiflowClusterMaster, name)

	bucket := b.buildAWSBucket(name, tags)
	err = k8s.Apply(ctx, b.Local, bucket)
	if err != nil {
		return deploy.Error(err)
	}

	// TODO(zhengjiajin): remove this when provider-aws can get reference ID
	if meta.GetExternalName(bucket) == "" {
		return deploy.Error(fmt.Errorf("bucket(%s) arn is empty", name))
	}

	// allow node role access the bucket.
	policy := b.buildBucketPolicy(bucket, spec.NodeRoleArn)
	err = k8s.Apply(ctx, b.Local, policy)
	if err != nil {
		return deploy.Error(err)
	}

	return deploy.Done()
}

// PostCreate implements deploy.Task interface.
func (b *Bucket) PostCreate(ctx context.Context) deploy.Result {
	name := cluster.FormatBucketName(b.Config.FullManage.Project, b.Network, b.Namer.GetTiflowClusterName())

	// check bucket
	bucket := &s3v1beta1.Bucket{
		ObjectMeta: metav1.ObjectMeta{
			Name: name,
		},
	}
	err := b.Local.Get(ctx, client.ObjectKeyFromObject(bucket), bucket)
	if err != nil {
		return deploy.Error(err)
	}

	ready := bucket.GetCondition(xpv1.TypeReady)
	synced := bucket.GetCondition(xpv1.TypeSynced)

	if ready.Status != corev1.ConditionTrue && synced.Status != corev1.ConditionTrue {
		cluster.UpdateBucketReadyForMaster(b.TiflowClusterMaster, false)
		return deploy.Error(fmt.Errorf("bucket not ready: %s", name))
	}

	// check policy
	bucketPolicy := &s3v1alpha3.BucketPolicy{
		ObjectMeta: metav1.ObjectMeta{
			Name: name,
		},
	}
	err = b.Local.Get(ctx, client.ObjectKeyFromObject(bucketPolicy), bucketPolicy)
	if err != nil {
		return deploy.Error(err)
	}
	ready = bucketPolicy.GetCondition(xpv1.TypeReady)
	synced = bucketPolicy.GetCondition(xpv1.TypeSynced)
	if ready.Status != corev1.ConditionTrue && synced.Status != corev1.ConditionTrue {
		cluster.UpdateBucketReadyForMaster(b.TiflowClusterMaster, false)
		return deploy.Error(fmt.Errorf("bucket policy not ready: %s", name))
	}

	cluster.UpdateBucketReadyForMaster(b.TiflowClusterMaster, true)
	return deploy.Done()
}

// Delete implements deploy.Task interface.
func (b *Bucket) Delete(ctx context.Context) deploy.Result {
	name := cluster.FormatBucketName(b.Config.FullManage.Project, b.Network, b.Namer.GetTiflowClusterName())
	b.Logger.Info("delete S3 bucket", "bucket", name)

	if err := b.emptyBucket(ctx, name); err != nil {
		return deploy.Error(err)
	}

	bucketPolicy := &s3v1alpha3.BucketPolicy{
		ObjectMeta: metav1.ObjectMeta{
			Name: name,
		},
	}
	err := k8s.DeleteObject(ctx, b.Local, bucketPolicy)
	if err != nil {
		return deploy.Error(err)
	}

	bucket := &s3v1beta1.Bucket{
		ObjectMeta: metav1.ObjectMeta{
			Name: name,
		},
	}
	err = k8s.DeleteObject(ctx, b.Local, bucket)
	if err != nil {
		return deploy.Error(err)
	}

	return deploy.Done()
}

func (b *Bucket) PostDelete(ctx context.Context) deploy.Result {
	name := cluster.FormatBucketName(b.Config.FullManage.Project, b.Network, b.Namer.GetTiflowClusterName())
	bucket := &s3v1beta1.Bucket{
		ObjectMeta: metav1.ObjectMeta{
			Name: name,
		},
	}
	if err := k8s.GetDeletedObject(ctx, b.Local, client.ObjectKeyFromObject(bucket), bucket); err != nil {
		return deploy.Error(err)
	}

	bucketPolicy := &s3v1alpha3.BucketPolicy{
		ObjectMeta: metav1.ObjectMeta{
			Name: name,
		},
	}
	if err := k8s.GetDeletedObject(ctx, b.Local, client.ObjectKeyFromObject(bucketPolicy), bucketPolicy); err != nil {
		return deploy.Error(err)
	}

	return deploy.Done()
}

func (b *Bucket) emptyBucket(ctx context.Context, bucket string) error {
	if bucket == "" {
		return nil
	}

	role, err := procedure.NetworkAssumeRole(b.Network)
	if err != nil {
		return err
	}

	client := b.AWSSVC.S3API(&procedure.AssumedRole{
		RoleARN:    role.RoleARN,
		ExternalID: role.ExternalID,
	})

	return procedure.EmptyBucket(ctx, client, bucket)
}

// example: https://github.com/crossplane-contrib/provider-aws/blob/master/examples/s3/bucket.yaml
func (b *Bucket) buildAWSBucket(name string, tags map[string]string) *s3v1beta1.Bucket {
	var tagging *s3v1beta1.Tagging
	if len(tags) > 0 {
		tagging = new(s3v1beta1.Tagging)
		tagging.TagSet = util.GetS3V1BetaTags(tags)
	}

	return &s3v1beta1.Bucket{
		ObjectMeta: metav1.ObjectMeta{
			Name: name,
		},
		Spec: s3v1beta1.BucketSpec{
			ResourceSpec: xpv1.ResourceSpec{
				DeletionPolicy:          xpv1.DeletionDelete,
				ProviderConfigReference: &xpv1.Reference{Name: namer.UniqueIdentifier(b.Network.ObjectMeta)},
			},
			ForProvider: s3v1beta1.BucketParameters{
				LocationConstraint: b.Network.Spec.Region,
				ObjectOwnership:    aws.String(string(types.ObjectOwnershipBucketOwnerEnforced)),
				BucketTagging:      tagging,
				PublicAccessBlockConfiguration: &s3v1beta1.PublicAccessBlockConfiguration{
					BlockPublicAcls:       aws.Bool(true),
					BlockPublicPolicy:     aws.Bool(true),
					IgnorePublicAcls:      aws.Bool(true),
					RestrictPublicBuckets: aws.Bool(true),
				},
				ServerSideEncryptionConfiguration: &s3v1beta1.ServerSideEncryptionConfiguration{
					Rules: []s3v1beta1.ServerSideEncryptionRule{
						{
							ApplyServerSideEncryptionByDefault: s3v1beta1.ServerSideEncryptionByDefault{
								SSEAlgorithm: "AES256",
							},
						},
					},
				},
			},
		},
	}
}

func (b *Bucket) buildBucketPolicy(bucket *s3v1beta1.Bucket, allowRoleARN string) *s3v1alpha3.BucketPolicy {
	bucketName := meta.GetExternalName(bucket)
	resource := fmt.Sprintf("arn:aws:s3:::%s", bucketName)

	return &s3v1alpha3.BucketPolicy{
		ObjectMeta: metav1.ObjectMeta{
			Name: bucket.Name,
		},
		Spec: s3v1alpha3.BucketPolicySpec{
			ResourceSpec: xpv1.ResourceSpec{
				DeletionPolicy:          xpv1.DeletionDelete,
				ProviderConfigReference: &xpv1.Reference{Name: namer.UniqueIdentifier(b.Network.ObjectMeta)},
			},
			Parameters: s3v1alpha3.BucketPolicyParameters{
				Region:     b.Network.Spec.Region,
				BucketName: pointer.String(bucketName),
				Policy: &s3common.BucketPolicyBody{
					Version: "2012-10-17",
					Statements: []s3common.BucketPolicyStatement{
						{
							Effect: "Allow",
							Principal: &s3common.BucketPrincipal{
								AWSPrincipals: []s3common.AWSPrincipal{
									{
										IAMRoleARN: pointer.String(allowRoleARN),
									},
								},
							},
							Action:   []string{"s3:*"},
							Resource: []string{resource, resource + "/*"},
						},
					},
				},
			},
		},
	}
}