package eks

import (
	"context"
	"fmt"
	"strings"

	"github.com/aws/smithy-go/ptr"
	"github.com/crossplane-contrib/provider-aws/apis/iam/v1beta1"
	xpv1 "github.com/crossplane/crossplane-runtime/apis/common/v1"
	"github.com/crossplane/crossplane-runtime/pkg/meta"
	"k8s.io/apimachinery/pkg/api/errors"
	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
	"sigs.k8s.io/controller-runtime/pkg/client"
	"sigs.k8s.io/controller-runtime/pkg/controller/controllerutil"
	"sigs.k8s.io/controller-runtime/pkg/log"

	"github.com/tidbcloud/eks-provider/controllers/cluster/task"
	cloudv1 "github.com/tidbcloud/infra-api/api/v1"
	"github.com/tidbcloud/infra-provider/pkg/deploy"
	"github.com/tidbcloud/infra-provider/pkg/k8s"
	"github.com/tidbcloud/infra-provider/pkg/util"
)

const (
	// Inline policy name suffix attached to the BatchOps role.
	batchOpsRolePolicyNameSuffix = "batchops-export"
	batchOpsRoleDescription      = "BatchOps export role for TiDB clusters"
	// Trust policy allowing the S3 Batch Operations service to assume the role.
	batchOpsTrustPolicy = `{
    "Version": "2012-10-17",
    "Statement": [
        {
            "Effect": "Allow",
            "Principal": {
                "Service": "batchoperations.s3.amazonaws.com"
            },
            "Action": "sts:AssumeRole"
        }
    ]
}`
	// batchOpsPolicyDocumentTmpl allows S3 Batch Operations to write to target S3 buckets.
	// NOTE:
	//   1. Source bucket access is granted by the dataflow-service backup export workflow.
	//   2. Target S3 buckets must be in a different AWS account than the source bucket.
	batchOpsPolicyDocumentTmpl = `{
    "Version": "2012-10-17",
    "Statement": [
        {
            "Sid": "AllowCrossAccountWrite",
            "Effect": "Allow",
            "Action": [
                "s3:PutObject",
                "s3:PutObjectAcl"
            ],
            "Resource": "arn:aws:s3:::*/*",
            "Condition": {
                "StringNotEquals": {
                    "s3:ResourceAccount": "%s"
                }
            }
        }
    ]
}`
)

// BatchOpsRole creates a dedicated IAM role for S3 Batch Operations based backup export.
// The role name follows dataflow-service convention: batchops-role-<projectID>-<clusterID>.
// It is cluster-scoped and deleted together with the cluster.
type BatchOpsRole struct {
	*task.BaseTask
}

func NewBatchOpsRole(ctx *task.Context) deploy.Task {
	return &BatchOpsRole{BaseTask: &task.BaseTask{Context: ctx}}
}

func (t *BatchOpsRole) Name() string {
	return "BatchOpsRole"
}

// Create ensures the BatchOps role, policy and attachment exist for the current cluster.
func (t *BatchOpsRole) Create(ctx context.Context) deploy.Result {

	logger := log.FromContext(ctx)

	// TODO CJL remove me after release finish
	if !t.Config.ClusterController.AWS.EnableBatchOpsExport {
		logger.Info("BatchOpsRole creation skipped as BatchOps export is disabled")
		return deploy.Done()
	}

	roleName := t.formatBatchOpsRole()

	accountID := t.Cluster.Spec.Kubernetes.Account
	if strings.TrimSpace(accountID) == "" {
		return deploy.Error(fmt.Errorf("cluster kubernetes.account is empty"))
	}

	policyDoc := buildBatchOpsPolicyDocument(accountID)

	role, roleReady, err := createCustomRole(ctx, t.Local, roleName, batchOpsTrustPolicy, t.GetAWSProviderConfigName(), t.Cluster.Labels)
	if err != nil {
		return deploy.Error(err)
	}

	policyARN, policyReady, err := createCustomPolicy(ctx, t.Local, roleName, batchOpsRolePolicyNameSuffix, batchOpsRoleDescription, policyDoc, t.GetAWSProviderConfigName())
	if err != nil {
		return deploy.Error(err)
	}

	if roleReady && policyReady {
		if err := attachPolicies(ctx, t.Local, role, []string{policyARN}, t.GetAWSProviderConfigName()); err != nil {
			return deploy.Error(err)
		}
	} else {
		return deploy.Error(fmt.Errorf("batchops role or policy is not ready"))
	}

	return deploy.Done()
}

// Delete removes the policy attachment, policy and role created for BatchOps.
func (t *BatchOpsRole) Delete(ctx context.Context) deploy.Result {
	roleName := t.formatBatchOpsRole()
	role := &v1beta1.Role{ObjectMeta: metav1.ObjectMeta{Name: roleName}}

	// Detach and delete policy
	policy := &v1beta1.Policy{ObjectMeta: metav1.ObjectMeta{Name: buildPolicyName(roleName, batchOpsRolePolicyNameSuffix)}}
	if err := t.detachPolicies(ctx, role, []string{policy.Name}); err != nil {
		return deploy.Error(err)
	}
	_ = client.IgnoreNotFound(t.Local.Delete(ctx, policy))

	// Delete role
	if err := client.IgnoreNotFound(t.Local.Delete(ctx, role)); err != nil {
		return deploy.Error(err)
	}

	return deploy.Done()
}

func (t *BatchOpsRole) formatBatchOpsRole() string {
	return fmt.Sprintf("batchops-role-%s-%s", t.Cluster.GetProjectID(), t.Cluster.GetClusterID())
}

// buildBatchOpsPolicyDocument fills the target-write policy with the source account guardrail.
func buildBatchOpsPolicyDocument(accountID string) string {
	return fmt.Sprintf(batchOpsPolicyDocumentTmpl, accountID)
}

// createCustomRole creates/updates the Crossplane IAM Role with the given trust policy.
func createCustomRole(ctx context.Context, cli client.Client, roleName, trustPolicy, providerConfig string, labels map[string]string) (*v1beta1.Role, bool, error) {
	role := &v1beta1.Role{
		TypeMeta: metav1.TypeMeta{
			APIVersion: "iam.aws.crossplane.io/v1beta1",
			Kind:       "Role",
		},
		ObjectMeta: metav1.ObjectMeta{
			Name: roleName,
		},
		Spec: v1beta1.RoleSpec{
			ForProvider: v1beta1.RoleParameters{
				Description:              ptr.String(batchOpsRoleDescription),
				AssumeRolePolicyDocument: trustPolicy,
				Tags:                     iamTagFromLabels(util.AddResourceName(cloudv1.TagsFromLabels(labels), roleName)),
			},
			ResourceSpec: xpv1.ResourceSpec{
				DeletionPolicy:          xpv1.DeletionDelete,
				ProviderConfigReference: &xpv1.Reference{Name: providerConfig},
			},
		},
	}

	if err := k8s.Apply(ctx, cli, role); err != nil {
		return nil, false, err
	}

	ready := role.Status.AtProvider.ARN != ""
	return role, ready, nil
}

// createCustomPolicy creates/updates the inline policy attached to the BatchOps role.
func createCustomPolicy(ctx context.Context, cli client.Client, roleName, suffix, description, document, providerConfig string) (policyARN string, ready bool, err error) {
	name := buildPolicyName(roleName, suffix)
	policy := v1beta1.Policy{
		ObjectMeta: metav1.ObjectMeta{
			Name:   name,
			Labels: map[string]string{RoleRefLabelKey: roleName},
		},
		Spec: v1beta1.PolicySpec{
			ForProvider: v1beta1.PolicyParameters{
				Description: ptr.String(description),
				Name:        name,
			},
			ResourceSpec: xpv1.ResourceSpec{
				DeletionPolicy:          xpv1.DeletionDelete,
				ProviderConfigReference: &xpv1.Reference{Name: providerConfig},
			},
		},
	}
	_, err = controllerutil.CreateOrUpdate(ctx, cli, &policy, func() error {
		policy.Spec.ForProvider.Document = *ptr.String(document)
		return nil
	})
	if err != nil {
		return "", false, err
	}
	if policy.Status.AtProvider.ARN == "" {
		return "", false, nil
	}
	return policy.Status.AtProvider.ARN, true, nil
}

func attachPolicies(ctx context.Context, cli client.Client, role *v1beta1.Role, policyARNs []string, providerConfig string) error {
	roleName := meta.GetExternalName(role)
	if roleName == "" {
		roleName = role.Name
	}
	for _, policyARN := range policyARNs {
		attachment := &v1beta1.RolePolicyAttachment{
			ObjectMeta: metav1.ObjectMeta{
				Name:   policyAttachmentName(role.Name, policyARN),
				Labels: map[string]string{RoleRefLabelKey: role.Name},
			},
			Spec: v1beta1.RolePolicyAttachmentSpec{
				ForProvider: v1beta1.RolePolicyAttachmentParameters{
					PolicyARN: policyARN,
					RoleName:  roleName,
				},
				ResourceSpec: xpv1.ResourceSpec{
					DeletionPolicy:          xpv1.DeletionDelete,
					ProviderConfigReference: &xpv1.Reference{Name: providerConfig},
				},
			},
		}
		if err := k8s.Apply(ctx, cli, attachment); err != nil {
			return err
		}
	}
	return nil
}

// detachPolicies removes role-policy attachments for the provided policy names.
func (t *BatchOpsRole) detachPolicies(ctx context.Context, role *v1beta1.Role, policyNames []string) error {
	for _, name := range policyNames {
		if name == "" {
			continue
		}
		cur := &v1beta1.Policy{}
		if err := t.Local.Get(ctx, client.ObjectKey{Name: name}, cur); err != nil {
			if errors.IsNotFound(err) {
				continue
			}
			return err
		}
		policyARN := cur.Status.AtProvider.ARN
		if policyARN == "" {
			continue
		}
		attachment := &v1beta1.RolePolicyAttachment{
			ObjectMeta: metav1.ObjectMeta{
				Name: policyAttachmentName(role.Name, policyARN),
			},
		}
		_ = client.IgnoreNotFound(t.Local.Delete(ctx, attachment))
	}
	return nil
}