package eks

import (
	"context"
	"fmt"
	"os"
	"time"

	"github.com/aws/aws-sdk-go/aws"
	"github.com/aws/aws-sdk-go/aws/awserr"
	"github.com/aws/aws-sdk-go/aws/credentials/stscreds"
	"github.com/aws/aws-sdk-go/aws/session"
	"github.com/aws/aws-sdk-go/service/s3"
	s3common "github.com/crossplane-contrib/provider-aws/apis/s3/common"
	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"
	"github.com/pingcap/errors"
	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"

	"github.com/tidbcloud/eks-provider/controllers/dm/task"
	"github.com/tidbcloud/infra-provider/pkg/deploy"
	"github.com/tidbcloud/infra-provider/pkg/k8s"
	"github.com/tidbcloud/infra-provider/pkg/util"
)

type Bucket struct {
	*task.BaseTask
}

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

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

// Create implements deploy.Task interface.
func (t *Bucket) Create(ctx context.Context) deploy.Result {
	if res := ensureDMNodeRoleARNReady(t.DMJob.Status.AWSNodeRoleARN); !res.IsDoneState() {
		return res
	}

	t.BeginAt = util.Pointer(time.Now())
	name := getBucketName(t.EnvProvider.GetAccount(), t.GetRegion(), t.GetMasterName())

	labels := t.GetLables()
	labels["dm_job_name"] = t.DMJob.Name

	bucket := t.buildS3Bucket(name, labels)
	err := k8s.CreateIfNotExist(ctx, t.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 externalName is empty: %s", name))
	}

	return deploy.Done()
}

// PostCreate implements deploy.Task interface.
func (t *Bucket) PostCreate(ctx context.Context) deploy.Result {
	name := getBucketName(t.EnvProvider.GetAccount(), t.GetRegion(), t.GetMasterName())

	bucket := &s3v1beta1.Bucket{
		ObjectMeta: metav1.ObjectMeta{
			Name: name,
		},
	}
	err := t.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 {
		return deploy.Error(fmt.Errorf("bucket not ready name: %s", name))
	}

	t.DMJob.Status.BlobContainerName = name
	return deploy.Done()
}

func (t *Bucket) emptyBucket(ctx context.Context, bucketName string) error {
	// get assume role arn
	assumeRoleArn := util.GetRoleARN(t.EnvProvider.GetAccount(), t.Config.FullManage.AssumeRoleName)

	sess, err := session.NewSession(&aws.Config{
		Region: aws.String(t.GetRegion()),
	})
	if err != nil {
		return fmt.Errorf("failed to create AWS session: %v", err)
	}

	// use assume role to create credentials
	creds := stscreds.NewCredentials(sess, assumeRoleArn, func(provider *stscreds.AssumeRoleProvider) {
		provider.ExternalID = aws.String(t.Config.FullManage.AssumeRoleExternalID)
	})

	// use assume role credentials to create s3 client
	svc := s3.New(sess, &aws.Config{
		Credentials: creds,
		Region:      aws.String(t.GetRegion()),
	})

	// Delete all objects
	err = svc.ListObjectsV2PagesWithContext(ctx, &s3.ListObjectsV2Input{
		Bucket: aws.String(bucketName),
	}, func(page *s3.ListObjectsV2Output, lastPage bool) bool {
		if len(page.Contents) == 0 {
			return false
		}

		objectIds := make([]*s3.ObjectIdentifier, 0, len(page.Contents))
		for _, obj := range page.Contents {
			objectIds = append(objectIds, &s3.ObjectIdentifier{
				Key: obj.Key,
			})
		}

		_, err = svc.DeleteObjectsWithContext(ctx, &s3.DeleteObjectsInput{
			Bucket: aws.String(bucketName),
			Delete: &s3.Delete{
				Objects: objectIds,
			},
		})
		if err != nil {
			t.Logger.Error(err, "failed to delete objects")
			return false
		}

		return !lastPage
	})

	if err != nil {
		return fmt.Errorf("failed to empty bucket: %v", err)
	}

	// Delete all object versions
	err = svc.ListObjectVersionsPagesWithContext(ctx, &s3.ListObjectVersionsInput{
		Bucket: aws.String(bucketName),
	}, func(page *s3.ListObjectVersionsOutput, lastPage bool) bool {
		objectIds := make([]*s3.ObjectIdentifier, 0)

		for _, version := range page.Versions {
			objectIds = append(objectIds, &s3.ObjectIdentifier{
				Key:       version.Key,
				VersionId: version.VersionId,
			})
		}

		for _, marker := range page.DeleteMarkers {
			objectIds = append(objectIds, &s3.ObjectIdentifier{
				Key:       marker.Key,
				VersionId: marker.VersionId,
			})
		}

		if len(objectIds) > 0 {
			_, err = svc.DeleteObjectsWithContext(ctx, &s3.DeleteObjectsInput{
				Bucket: aws.String(bucketName),
				Delete: &s3.Delete{
					Objects: objectIds,
				},
			})
			if err != nil {
				t.Logger.Error(err, "failed to delete objects versions")
				return false
			}
		}

		return !lastPage
	})

	if err != nil {
		t.Logger.Error(err, "failed to empty bucket versions")
		return fmt.Errorf("failed to empty bucket versions: %v", err)
	}

	return nil
}

func (t *Bucket) isNotExistError(err error) bool {
	if err == nil {
		return false
	}
	err = errors.Cause(err)
	if os.IsNotExist(err) {
		return true
	}
	if aerr, ok := err.(awserr.Error); ok {
		switch aerr.Code() {
		case s3.ErrCodeNoSuchBucket, s3.ErrCodeNoSuchKey, "NotFound":
			return true
		}
	}
	return false
}

// Delete implements deploy.Task interface.
func (t *Bucket) Delete(ctx context.Context) deploy.Result {
	name := getBucketName(t.EnvProvider.GetAccount(), t.GetRegion(), t.GetMasterName())

	// Check if bucket exists before trying to empty it
	// get assume role arn
	assumeRoleArn := util.GetRoleARN(t.EnvProvider.GetAccount(), t.Config.FullManage.AssumeRoleName)

	sess, err := session.NewSession(&aws.Config{
		Region: aws.String(t.GetRegion()),
	})
	if err != nil {
		return deploy.Error(fmt.Errorf("failed to create AWS session: %v", err))
	}

	// use assume role to create credentials
	creds := stscreds.NewCredentials(sess, assumeRoleArn, func(provider *stscreds.AssumeRoleProvider) {
		provider.ExternalID = aws.String(t.Config.FullManage.AssumeRoleExternalID)
	})

	// use assume role credentials to create s3 client
	svc := s3.New(sess, &aws.Config{
		Credentials: creds,
		Region:      aws.String(t.GetRegion()),
	})

	_, err = svc.HeadBucketWithContext(ctx, &s3.HeadBucketInput{
		Bucket: aws.String(name),
	})

	if err != nil {
		if t.isNotExistError(err) {
			t.Logger.Info("bucket not found, skip empty bucket & delete bucket")
			return deploy.Done()
		}

		t.Logger.Error(err, "failed to check bucket")
		return deploy.Error(err)
	}

	// Bucket exists, try to empty it
	if err := t.emptyBucket(ctx, name); err != nil {
		t.Logger.Error(err, "failed to empty bucket")
		return deploy.Error(err)
	}

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

	return deploy.Done()
}

func (t *Bucket) PostDelete(ctx context.Context) deploy.Result {
	name := getBucketName(t.EnvProvider.GetAccount(), t.GetRegion(), t.GetMasterName())
	bucket := &s3v1beta1.Bucket{ObjectMeta: metav1.ObjectMeta{
		Name: name,
	}}
	if err := k8s.GetDeletedObject(ctx, t.Local, client.ObjectKeyFromObject(bucket), bucket); err != nil {
		return deploy.Error(err)
	}

	return deploy.Done()
}

func (t *Bucket) buildS3Bucket(name string, labels map[string]string) *s3v1beta1.Bucket {
	var tagging *s3v1beta1.Tagging
	if len(labels) > 0 {
		tagging = new(s3v1beta1.Tagging)
		tagging.TagSet = util.GetS3V1BetaTags(labels)
	}

	providerConfigRef := t.GetProviderConfigName()
	nodeRoleARN := t.DMJob.Status.AWSNodeRoleARN

	return &s3v1beta1.Bucket{
		ObjectMeta: metav1.ObjectMeta{
			Name:   name,
			Labels: labels,
		},
		Spec: s3v1beta1.BucketSpec{
			ResourceSpec: xpv1.ResourceSpec{
				DeletionPolicy:          xpv1.DeletionDelete,
				ProviderConfigReference: &xpv1.Reference{Name: providerConfigRef},
				WriteConnectionSecretToReference: &xpv1.SecretReference{
					Name:      name,
					Namespace: t.DMJob.Namespace,
				},
			},
			ForProvider: s3v1beta1.BucketParameters{
				LocationConstraint: t.GetRegion(),
				ObjectOwnership:    aws.String("BucketOwnerEnforced"),
				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",
							},
						},
					},
				},
				Policy: &s3common.BucketPolicyBody{
					Version: "2012-10-17",
					Statements: []s3common.BucketPolicyStatement{
						{
							Effect: "Allow",
							Principal: &s3common.BucketPrincipal{
								AWSPrincipals: []s3common.AWSPrincipal{
									{
										IAMRoleARN: pointer.String(nodeRoleARN),
									},
								},
							},
							Action:   []string{"s3:*"},
							Resource: []string{fmt.Sprintf("arn:aws:s3:::%s", name), fmt.Sprintf("arn:aws:s3:::%s/*", name)},
						},
					},
				},
			},
		},
	}
}

func getBucketName(account string, region string, master string) string {
	const (
		labelLengthLimit = 63
		hashSize         = 8
		resultLimit      = labelLengthLimit - hashSize - 1
	)

	result := fmt.Sprintf("dm-%s-%s-%s", account, region, util.FnvHash(master))
	if len(result) > resultLimit {
		result = result[:resultLimit]
	}

	return result
}