package keyspacerole

import (
	"context"
	"fmt"

	xpv1 "github.com/crossplane/crossplane-runtime/apis/common/v1"
	dmpbv1 "github.com/tidbcloud/cluster-service-ng/api/tidbcloud/dataplane-manager/v1"
	bizerr "github.com/tidbcloud/infra-provider/regional-server/pkg/common/errors"
	"github.com/tidbcloud/infra-provider/regional-server/pkg/controllers"
	"github.com/tidbcloud/infra-provider/regional-server/pkg/k8s"
	"github.com/tidbcloud/infra-provider/regional-server/pkg/util"
	iamv1beta1 "github.com/upbound/provider-aws/apis/iam/v1beta1"
	apierrors "k8s.io/apimachinery/pkg/api/errors"
	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
	"k8s.io/apimachinery/pkg/types"
	"k8s.io/utils/ptr"
)

func (p awsKeyspaceRoleProvider) apply(ctx context.Context, spec *keyspaceRoleSpec) error {
	// Keyspace roles and policies are separate Crossplane resources; retry completes any partial apply.
	if err := p.validateExistingOwners(ctx, spec); err != nil {
		return err
	}
	if err := p.applyRole(ctx, spec); err != nil {
		return bizerr.Wrap(bizerr.K8SClientError, "apply keyspace role error", err)
	}
	if err := p.applyRolePolicy(ctx, spec); err != nil {
		return bizerr.Wrap(bizerr.K8SClientError, "apply keyspace role policy error", err)
	}
	return nil
}

func (p awsKeyspaceRoleProvider) validateExistingOwners(ctx context.Context, spec *keyspaceRoleSpec) error {
	if err := validateExistingResourceOwner(ctx, p.local, &iamv1beta1.Role{}, spec.roleName, spec.dataplaneID); err != nil {
		return err
	}
	return validateExistingResourceOwner(ctx, p.local, &iamv1beta1.RolePolicy{}, spec.rolePolicyName, spec.dataplaneID)
}

func (p awsKeyspaceRoleProvider) applyRole(ctx context.Context, spec *keyspaceRoleSpec) error {
	trustPolicy, err := buildTrustPolicy(spec.accountID, spec.assumeRoleARN, spec.externalID)
	if err != nil {
		return err
	}
	role := &iamv1beta1.Role{
		ObjectMeta: metav1.ObjectMeta{
			Name:        spec.roleName,
			Labels:      spec.labels,
			Annotations: spec.annotations,
		},
		Spec: iamv1beta1.RoleSpec{
			ForProvider: iamv1beta1.RoleParameters{
				Description:      ptr.To(fmt.Sprintf("Keyspace role for keyspace %s", spec.keyspaceName)),
				AssumeRolePolicy: ptr.To(trustPolicy),
				Tags:             util.ToMapPointer(spec.cloudTags),
			},
			ResourceSpec: xpv1.ResourceSpec{
				ProviderConfigReference: &xpv1.Reference{Name: spec.providerConfig},
				ManagementPolicies:      controllers.GetResourceManagementPolicies(),
			},
		},
	}
	return k8s.Apply(ctx, p.local, role)
}

func (p awsKeyspaceRoleProvider) applyRolePolicy(ctx context.Context, spec *keyspaceRoleSpec) error {
	policy, err := buildS3Policy(spec.bucketName, spec.allowedPrefix)
	if err != nil {
		return err
	}
	rolePolicy := &iamv1beta1.RolePolicy{
		ObjectMeta: metav1.ObjectMeta{
			Name:        spec.rolePolicyName,
			Labels:      spec.labels,
			Annotations: spec.annotations,
		},
		Spec: iamv1beta1.RolePolicySpec{
			ForProvider: iamv1beta1.RolePolicyParameters{
				Policy:  ptr.To(policy),
				RoleRef: &xpv1.Reference{Name: spec.roleName},
			},
			ResourceSpec: xpv1.ResourceSpec{
				ProviderConfigReference: &xpv1.Reference{Name: spec.providerConfig},
				ManagementPolicies:      controllers.GetResourceManagementPolicies(),
			},
		},
	}
	return k8s.Apply(ctx, p.local, rolePolicy)
}

func (p awsKeyspaceRoleProvider) get(ctx context.Context, dataplaneID, keyspaceName, roleName string) (*dmpbv1.KeyspaceRole, error) {
	role := &iamv1beta1.Role{}
	if err := p.local.Get(ctx, types.NamespacedName{Name: roleName}, role); err != nil {
		if apierrors.IsNotFound(err) {
			return nil, bizerr.New(bizerr.ResourceNotFound, "keyspace role not found")
		}
		return nil, bizerr.Wrap(bizerr.K8SClientError, "get keyspace role error", err)
	}
	if err := validateDataplaneOwner(role.Labels, dataplaneID); err != nil {
		return nil, err
	}

	rolePolicy := &iamv1beta1.RolePolicy{}
	if err := p.local.Get(ctx, types.NamespacedName{Name: rolePolicyName(roleName)}, rolePolicy); err != nil {
		if apierrors.IsNotFound(err) {
			return keyspaceRoleResponse(dataplaneID, keyspaceName, awsRoleARN(role), role.Labels, role.Annotations, dmpbv1.KeyspaceRole_Creating, "keyspace role policy is not ready"), nil
		}
		return nil, bizerr.Wrap(bizerr.K8SClientError, "get keyspace role policy error", err)
	}

	status, message := keyspaceRoleStatus(awsRoleARN(role), managedResource{
		name:   "keyspace role",
		object: role,
		status: role.Status.ResourceStatus,
	}, managedResource{
		name:   "keyspace role policy",
		object: rolePolicy,
		status: rolePolicy.Status.ResourceStatus,
	})
	return keyspaceRoleResponse(dataplaneID, keyspaceName, awsRoleARN(role), role.Labels, role.Annotations, status, message), nil
}

func (p awsKeyspaceRoleProvider) delete(ctx context.Context, dataplaneID, roleName string) error {
	role := &iamv1beta1.Role{}
	if err := p.local.Get(ctx, types.NamespacedName{Name: roleName}, role); err != nil {
		if !apierrors.IsNotFound(err) {
			return bizerr.Wrap(bizerr.K8SClientError, "get keyspace role error", err)
		}
	} else if err := validateDataplaneOwner(role.Labels, dataplaneID); err != nil {
		return err
	}

	if err := controllers.DeleteUpboundResource(ctx, p.local, &iamv1beta1.RolePolicy{
		ObjectMeta: metav1.ObjectMeta{Name: rolePolicyName(roleName)},
	}); err != nil {
		return bizerr.Wrap(bizerr.K8SClientError, "delete keyspace role policy error", err)
	}
	if err := controllers.DeleteUpboundResource(ctx, p.local, &iamv1beta1.Role{
		ObjectMeta: metav1.ObjectMeta{Name: roleName},
	}); err != nil {
		return bizerr.Wrap(bizerr.K8SClientError, "delete keyspace role error", err)
	}
	return nil
}

func buildTrustPolicy(accountID, principalARN, externalID string) (string, error) {
	if accountID == "" || principalARN == "" || externalID == "" {
		return "", bizerr.New(bizerr.ParameterInvalid, "account id, principal arn and external id are required")
	}
	policy := map[string]any{
		"Version": "2012-10-17",
		"Statement": []map[string]any{
			{
				"Effect": "Allow",
				"Principal": map[string]string{
					"AWS": fmt.Sprintf("arn:aws:iam::%s:root", accountID),
				},
				"Action": "sts:AssumeRole",
				"Condition": map[string]map[string]string{
					"ArnEquals": {
						"aws:PrincipalArn": principalARN,
					},
					"StringEquals": {
						"sts:ExternalId": externalID,
					},
				},
			},
		},
	}
	return marshalPolicy(policy)
}

func buildS3Policy(bucketName, allowedPrefix string) (string, error) {
	allowedPrefix = normalizeS3Prefix(allowedPrefix)
	if bucketName == "" || allowedPrefix == "" {
		return "", fmt.Errorf("bucket name and allowed prefix are required")
	}
	policy := map[string]any{
		"Version": "2012-10-17",
		"Statement": []map[string]any{
			{
				"Sid":      "ListOwnKeyspacePrefixOnly",
				"Effect":   "Allow",
				"Action":   "s3:ListBucket",
				"Resource": fmt.Sprintf("arn:aws:s3:::%s", bucketName),
				"Condition": map[string]map[string][]string{
					"StringLike": {
						"s3:prefix": {
							allowedPrefix + "/*",
						},
					},
				},
			},
			{
				"Sid":    "ReadWriteOwnKeyspacePrefixOnly",
				"Effect": "Allow",
				"Action": []string{
					"s3:GetObject",
					"s3:PutObject",
					"s3:DeleteObject",
					"s3:AbortMultipartUpload",
					"s3:ListMultipartUploadParts",
				},
				"Resource": fmt.Sprintf("arn:aws:s3:::%s/%s/*", bucketName, allowedPrefix),
			},
		},
	}
	return marshalPolicy(policy)
}

func awsRoleARN(role *iamv1beta1.Role) string {
	if role.Status.AtProvider.Arn == nil {
		return ""
	}
	return *role.Status.AtProvider.Arn
}