package keyspacerole

import (
	"context"
	"encoding/json"
	"strings"
	"testing"

	ramv1alpha1 "github.com/crossplane-contrib/provider-upjet-alibabacloud/apis/ram/v1alpha1"
	xpv1 "github.com/crossplane/crossplane-runtime/apis/common/v1"
	"github.com/stretchr/testify/require"
	dmpbv1 "github.com/tidbcloud/cluster-service-ng/api/tidbcloud/dataplane-manager/v1"
	"github.com/tidbcloud/infra-provider/regional-server/api/v1alpha1"
	bizerr "github.com/tidbcloud/infra-provider/regional-server/pkg/common/errors"
	"github.com/tidbcloud/infra-provider/regional-server/pkg/config"
	"github.com/tidbcloud/infra-provider/regional-server/pkg/consts"
	"github.com/tidbcloud/infra-provider/regional-server/pkg/repository"
	dpmscheme "github.com/tidbcloud/infra-provider/regional-server/pkg/scheme"
	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"
	"sigs.k8s.io/controller-runtime/pkg/client"
	"sigs.k8s.io/controller-runtime/pkg/client/fake"
)

func TestCreateKeyspaceRoleAppliesRoleAndPolicy(t *testing.T) {
	ctx := context.Background()
	cli := newFakeClient(
		testDataPlane(),
		testAPPBox(),
		testBucket(v1alpha1.BucketPhaseReady),
	)
	svc := newTestService(cli)

	resp, err := svc.CreateKeyspaceRole(ctx, &dmpbv1.CreateKeyspaceRoleRequest{
		DataplaneId: "dp-1",
		KeyspaceRole: &dmpbv1.KeyspaceRole{
			KeyspaceName:  "10000001",
			BucketId:      "bucket-1",
			AllowedPrefix: "/10000001/",
			ExternalId:    "10000001",
			Tags: map[string]string{
				consts.TenantLabelKey:     "tenant-1",
				consts.SharedPoolLabelKey: "pool-1",
				consts.DataplaneLabelKey:  "overridden-dp",
				consts.BucketLabelKey:     "overridden-bucket",
				keyspaceNameLabelKey:      "overridden-role",
			},
		},
	})
	require.NoError(t, err)
	require.Equal(t, dmpbv1.KeyspaceRole_Creating, resp.Status)
	require.Equal(t, "bucket-1", resp.BucketId)
	require.Equal(t, "10000001", resp.AllowedPrefix)
	require.Equal(t, "10000001", resp.ExternalId)

	var role iamv1beta1.Role
	require.NoError(t, cli.Get(ctx, types.NamespacedName{Name: "test-keyspace-10000001"}, &role))
	require.Equal(t, "provider-dp", role.Spec.ResourceSpec.ProviderConfigReference.Name)
	require.Equal(t, "10000001", role.Labels[keyspaceNameLabelKey])
	require.Equal(t, "bucket-1", role.Labels[consts.BucketLabelKey])
	require.Equal(t, "dp-1", role.Labels[consts.DataplaneLabelKey])
	require.Equal(t, "pool-1", role.Labels[consts.SharedPoolLabelKey])
	require.Equal(t, "10000001", ptr.Deref(role.Spec.ForProvider.Tags["keyspace"], ""))
	require.Equal(t, "bucket-1", ptr.Deref(role.Spec.ForProvider.Tags["bucket-id"], ""))
	require.Equal(t, "dp-1", ptr.Deref(role.Spec.ForProvider.Tags["dataplane-id"], ""))
	require.Equal(t, "test-keyspace-10000001", ptr.Deref(role.Spec.ForProvider.Tags["Name"], ""))

	trustPolicy := decodePolicy(t, ptr.Deref(role.Spec.ForProvider.AssumeRolePolicy, ""))
	statement := trustPolicy["Statement"].([]any)[0].(map[string]any)
	require.Equal(t, "arn:aws:iam::123456789012:root", statement["Principal"].(map[string]any)["AWS"])
	condition := statement["Condition"].(map[string]any)
	require.Equal(t, "arn:aws:iam::123456789012:role/node-role", condition["ArnEquals"].(map[string]any)["aws:PrincipalArn"])
	require.Equal(t, "10000001", condition["StringEquals"].(map[string]any)["sts:ExternalId"])

	var rolePolicy iamv1beta1.RolePolicy
	require.NoError(t, cli.Get(ctx, types.NamespacedName{Name: "test-keyspace-10000001-policy"}, &rolePolicy))
	require.Equal(t, "test-keyspace-10000001", rolePolicy.Spec.ForProvider.RoleRef.Name)
	policy := ptr.Deref(rolePolicy.Spec.ForProvider.Policy, "")
	require.Contains(t, policy, "arn:aws:s3:::shared-bucket/10000001/*")
	requirePolicyListPrefixes(t, policy, "s3:prefix", []string{"10000001/*"})
	require.NotContains(t, policy, "s3:ListBucketMultipartUploads")
}

func TestCreateNonBYOCKeyspaceRoleWithEmptyImportRoleNameKeepsDynamicBehavior(t *testing.T) {
	ctx := context.Background()
	cli := newFakeClient(
		testDataPlane(),
		testAPPBox(),
		testBucket(v1alpha1.BucketPhaseReady),
	)
	svc := newTestServiceWithConfig(cli, config.RootConfig{
		Namespace:          "default",
		ResourceNamePrefix: "test",
		ControllerConfig: config.ControllerConfig{
			Provider: consts.ProviderTypeEKS,
		},
	})

	resp, err := svc.CreateKeyspaceRole(ctx, &dmpbv1.CreateKeyspaceRoleRequest{
		DataplaneId: "dp-1",
		KeyspaceRole: &dmpbv1.KeyspaceRole{
			KeyspaceName:  "10000001",
			BucketId:      "bucket-1",
			AllowedPrefix: "10000001",
			ExternalId:    "10000001",
		},
	})
	require.NoError(t, err)
	require.Equal(t, dmpbv1.KeyspaceRole_Creating, resp.Status)

	var role iamv1beta1.Role
	require.NoError(t, cli.Get(ctx, types.NamespacedName{Name: "test-keyspace-10000001"}, &role))
	var rolePolicy iamv1beta1.RolePolicy
	require.NoError(t, cli.Get(ctx, types.NamespacedName{Name: "test-keyspace-10000001-policy"}, &rolePolicy))
}

func TestCreateACKKeyspaceRoleAppliesRolePolicyAndAttachment(t *testing.T) {
	ctx := context.Background()
	cli := newFakeClient(
		testDataPlane(),
		testAPPBoxWithProvider(consts.ProviderTypeACK),
		testBucket(v1alpha1.BucketPhaseReady),
	)
	svc := newTestServiceWithProvider(cli, consts.ProviderTypeACK)

	resp, err := svc.CreateKeyspaceRole(ctx, &dmpbv1.CreateKeyspaceRoleRequest{
		DataplaneId: "dp-1",
		KeyspaceRole: &dmpbv1.KeyspaceRole{
			KeyspaceName:  "10000001",
			BucketId:      "bucket-1",
			AllowedPrefix: "/10000001/",
			ExternalId:    "10000001",
			Tags: map[string]string{
				consts.TenantLabelKey:     "tenant-1",
				consts.SharedPoolLabelKey: "pool-1",
			},
		},
	})
	require.NoError(t, err)
	require.Equal(t, dmpbv1.KeyspaceRole_Creating, resp.Status)

	var role ramv1alpha1.Role
	require.NoError(t, cli.Get(ctx, types.NamespacedName{Name: "test-keyspace-10000001"}, &role))
	require.Equal(t, "provider-dp", role.Spec.ResourceSpec.ProviderConfigReference.Name)
	require.Equal(t, "test-keyspace-10000001", ptr.Deref(role.Spec.ForProvider.RoleName, ""))
	require.Equal(t, "10000001", role.Labels[keyspaceNameLabelKey])
	require.Equal(t, "bucket-1", role.Labels[consts.BucketLabelKey])
	require.Equal(t, "pool-1", role.Labels[consts.SharedPoolLabelKey])

	trustPolicy := decodePolicy(t, ptr.Deref(role.Spec.ForProvider.AssumeRolePolicyDocument, ""))
	statement := trustPolicy["Statement"].([]any)[0].(map[string]any)
	require.Equal(t, "acs:ram::123456789012:root", statement["Principal"].(map[string]any)["RAM"].([]any)[0])
	condition := statement["Condition"].(map[string]any)
	require.Equal(t, "acs:ram::123456789012:role/noderole", condition["StringEquals"].(map[string]any)["acs:PrincipalARN"])
	require.Equal(t, "10000001", condition["StringEquals"].(map[string]any)["sts:ExternalId"])

	var policy ramv1alpha1.Policy
	require.NoError(t, cli.Get(ctx, types.NamespacedName{Name: "test-keyspace-10000001-policy"}, &policy))
	require.Equal(t, "test-keyspace-10000001-policy", ptr.Deref(policy.Spec.ForProvider.PolicyName, ""))
	policyDocument := ptr.Deref(policy.Spec.ForProvider.PolicyDocument, "")
	require.Contains(t, policyDocument, "acs:oss:*:*:shared-bucket/10000001/*")
	requirePolicyListPrefixes(t, policyDocument, "oss:Prefix", []string{"10000001/*"})
	require.Contains(t, policyDocument, "oss:ListParts")
	require.NotContains(t, policyDocument, "oss:ListMultipartUploads")

	var attachment ramv1alpha1.RolePolicyAttachment
	require.NoError(t, cli.Get(ctx, types.NamespacedName{Name: "test-keyspace-10000001-policy-attachment"}, &attachment))
	require.Equal(t, "Custom", ptr.Deref(attachment.Spec.ForProvider.PolicyType, ""))
	require.Equal(t, "test-keyspace-10000001-policy", attachment.Spec.ForProvider.PolicyNameRef.Name)
	require.Equal(t, "test-keyspace-10000001", ptr.Deref(attachment.Spec.ForProvider.RoleName, ""))
}

func TestCreateStaticBYOCKeyspaceRoleReturnsReadyWithoutCloudResources(t *testing.T) {
	ctx := context.Background()
	cli := newFakeClient(
		testDataPlane(),
		testBucket(v1alpha1.BucketPhaseReady),
	)
	svc := newTestStaticImportRoleService(cli, consts.ProviderTypeEKS)

	resp, err := svc.CreateKeyspaceRole(ctx, &dmpbv1.CreateKeyspaceRoleRequest{
		DataplaneId: "dp-1",
		KeyspaceRole: &dmpbv1.KeyspaceRole{
			KeyspaceName:  "10000001",
			BucketId:      "bucket-1",
			AllowedPrefix: "/10000001/",
			ExternalId:    "10000001",
			Tags:          map[string]string{consts.TenantLabelKey: "tenant-1"},
		},
	})
	require.NoError(t, err)
	require.Equal(t, dmpbv1.KeyspaceRole_Ready, resp.Status)
	require.Equal(t, "arn:aws:iam::123456789012:role/tidbcloud-import-role", resp.RoleArn)
	require.Equal(t, "bucket-1", resp.BucketId)
	require.Equal(t, "10000001", resp.AllowedPrefix)
	require.Equal(t, "10000001", resp.ExternalId)

	err = cli.Get(ctx, types.NamespacedName{Name: "test-keyspace-10000001"}, &iamv1beta1.Role{})
	require.True(t, apierrors.IsNotFound(err))
	err = cli.Get(ctx, types.NamespacedName{Name: "test-keyspace-10000001-policy"}, &iamv1beta1.RolePolicy{})
	require.True(t, apierrors.IsNotFound(err))
}

func TestCreateStaticBYOCKeyspaceRoleRejectsInvalidExternalID(t *testing.T) {
	for _, tc := range []struct {
		name       string
		keyspace   string
		externalID string
		message    string
	}{
		{
			name:       "missing keyspace name",
			externalID: "10000001",
			message:    "keyspace_role.keyspace_name is required",
		},
		{
			name:     "missing external id",
			keyspace: "10000001",
			message:  "keyspace_role.external_id is required",
		},
	} {
		t.Run(tc.name, func(t *testing.T) {
			svc := newTestStaticImportRoleService(newFakeClient(
				testDataPlane(),
				testBucket(v1alpha1.BucketPhaseReady),
			), consts.ProviderTypeEKS)

			resp, err := svc.CreateKeyspaceRole(context.Background(), &dmpbv1.CreateKeyspaceRoleRequest{
				DataplaneId: "dp-1",
				KeyspaceRole: &dmpbv1.KeyspaceRole{
					KeyspaceName:  tc.keyspace,
					BucketId:      "bucket-1",
					AllowedPrefix: "10000001",
					ExternalId:    tc.externalID,
				},
			})
			require.Nil(t, resp)
			require.True(t, bizerr.Is(err, bizerr.ParameterInvalid))
			require.ErrorContains(t, err, tc.message)
		})
	}
}

func TestCreateStaticBYOCKeyspaceRoleAllowsSeparateExternalID(t *testing.T) {
	ctx := context.Background()
	svc := newTestStaticImportRoleService(newFakeClient(
		testDataPlane(),
		testBucket(v1alpha1.BucketPhaseReady),
	), consts.ProviderTypeEKS)

	resp, err := svc.CreateKeyspaceRole(ctx, &dmpbv1.CreateKeyspaceRoleRequest{
		DataplaneId: "dp-1",
		KeyspaceRole: &dmpbv1.KeyspaceRole{
			KeyspaceName:  "10000001",
			BucketId:      "bucket-1",
			AllowedPrefix: "10000001",
			ExternalId:    "SYSTEM",
		},
	})
	require.NoError(t, err)
	require.Equal(t, dmpbv1.KeyspaceRole_Ready, resp.Status)
	require.Equal(t, "10000001", resp.KeyspaceName)
	require.Equal(t, "SYSTEM", resp.ExternalId)
	require.Equal(t, "10000001", resp.AllowedPrefix)
}

func TestGetStaticBYOCKeyspaceRoleReady(t *testing.T) {
	ctx := context.Background()
	svc := newTestStaticImportRoleService(newFakeClient(
		testDataPlane(),
		testBucket(v1alpha1.BucketPhaseReady),
	), consts.ProviderTypeEKS)

	resp, err := svc.GetKeyspaceRole(ctx, &dmpbv1.GetKeyspaceRoleRequest{
		DataplaneId:   "dp-1",
		KeyspaceName:  "10000001",
		BucketId:      "bucket-1",
		AllowedPrefix: "/10000001/",
	})
	require.NoError(t, err)
	require.Equal(t, dmpbv1.KeyspaceRole_Ready, resp.Status)
	require.Equal(t, "arn:aws:iam::123456789012:role/tidbcloud-import-role", resp.RoleArn)
	require.Equal(t, "bucket-1", resp.BucketId)
	require.Equal(t, "10000001", resp.AllowedPrefix)
	require.Equal(t, "10000001", resp.ExternalId)
}

func TestGetACKStaticBYOCKeyspaceRoleReady(t *testing.T) {
	ctx := context.Background()
	svc := newTestStaticImportRoleService(newFakeClient(
		testDataPlane(),
		testBucket(v1alpha1.BucketPhaseReady),
	), consts.ProviderTypeACK)

	resp, err := svc.GetKeyspaceRole(ctx, &dmpbv1.GetKeyspaceRoleRequest{
		DataplaneId:   "dp-1",
		KeyspaceName:  "10000001",
		BucketId:      "bucket-1",
		AllowedPrefix: "10000001",
	})
	require.NoError(t, err)
	require.Equal(t, dmpbv1.KeyspaceRole_Ready, resp.Status)
	require.Equal(t, "acs:ram::123456789012:role/tidbcloud-import-role", resp.RoleArn)
	require.Equal(t, "10000001", resp.ExternalId)
}

func TestGetStaticBYOCSystemKeyspaceRoleReady(t *testing.T) {
	ctx := context.Background()
	svc := newTestStaticImportRoleService(newFakeClient(
		testDataPlane(),
		testBucket(v1alpha1.BucketPhaseReady),
	), consts.ProviderTypeEKS)

	resp, err := svc.GetKeyspaceRole(ctx, &dmpbv1.GetKeyspaceRoleRequest{
		DataplaneId:   "dp-1",
		KeyspaceName:  "10000001",
		BucketId:      "bucket-1",
		AllowedPrefix: "10000001",
		ExternalId:    "SYSTEM",
	})
	require.NoError(t, err)
	require.Equal(t, dmpbv1.KeyspaceRole_Ready, resp.Status)
	require.Equal(t, "10000001", resp.KeyspaceName)
	require.Equal(t, "SYSTEM", resp.ExternalId)
	require.Equal(t, "10000001", resp.AllowedPrefix)
}

func TestDeleteStaticBYOCKeyspaceRoleIsNoop(t *testing.T) {
	ctx := context.Background()
	roleName := resourceName("test", "10000001")
	role := &iamv1beta1.Role{ObjectMeta: metav1.ObjectMeta{
		Name:   roleName,
		Labels: map[string]string{consts.DataplaneLabelKey: "dp-1"},
	}}
	rolePolicy := &iamv1beta1.RolePolicy{ObjectMeta: metav1.ObjectMeta{Name: rolePolicyName(roleName)}}
	cli := newFakeClient(role, rolePolicy)
	svc := newTestStaticImportRoleService(cli, consts.ProviderTypeEKS)

	resp, err := svc.DeleteKeyspaceRole(ctx, &dmpbv1.DeleteKeyspaceRoleRequest{
		DataplaneId:  "dp-1",
		KeyspaceName: "10000001",
	})
	require.NoError(t, err)
	require.NotNil(t, resp)
	require.NoError(t, cli.Get(ctx, types.NamespacedName{Name: roleName}, &iamv1beta1.Role{}))
	require.NoError(t, cli.Get(ctx, types.NamespacedName{Name: rolePolicyName(roleName)}, &iamv1beta1.RolePolicy{}))
}

func TestCreateBYOCKeyspaceRoleWithoutImportRoleNameIsNoop(t *testing.T) {
	ctx := context.Background()
	cli := newFakeClient(
		testDataPlane(),
		testAPPBox(),
		testBucket(v1alpha1.BucketPhaseReady),
	)
	svc := newTestBYOCServiceWithoutImportRole(cli, consts.ProviderTypeEKS)

	resp, err := svc.CreateKeyspaceRole(ctx, &dmpbv1.CreateKeyspaceRoleRequest{
		DataplaneId: "dp-1",
		KeyspaceRole: &dmpbv1.KeyspaceRole{
			KeyspaceName:  "10000001",
			BucketId:      "bucket-1",
			AllowedPrefix: "10000001",
			ExternalId:    "10000001",
		},
	})
	require.NoError(t, err)
	require.Equal(t, dmpbv1.KeyspaceRole_Failed, resp.Status)
	require.Equal(t, "byoc.importRoleName is not configured", resp.Message)

	err = cli.Get(ctx, types.NamespacedName{Name: "test-keyspace-10000001"}, &iamv1beta1.Role{})
	require.True(t, apierrors.IsNotFound(err))
	err = cli.Get(ctx, types.NamespacedName{Name: "test-keyspace-10000001-policy"}, &iamv1beta1.RolePolicy{})
	require.True(t, apierrors.IsNotFound(err))
}

func TestGetBYOCKeyspaceRoleWithoutImportRoleNameReturnsNotConfiguredStatus(t *testing.T) {
	ctx := context.Background()
	svc := newTestBYOCServiceWithoutImportRole(newFakeClient(
		testDataPlane(),
		testBucket(v1alpha1.BucketPhaseReady),
	), consts.ProviderTypeEKS)

	resp, err := svc.GetKeyspaceRole(ctx, &dmpbv1.GetKeyspaceRoleRequest{
		DataplaneId:   "dp-1",
		KeyspaceName:  "10000001",
		BucketId:      "bucket-1",
		AllowedPrefix: "10000001",
	})
	require.NoError(t, err)
	require.Equal(t, dmpbv1.KeyspaceRole_Failed, resp.Status)
	require.Equal(t, "byoc.importRoleName is not configured", resp.Message)
}

func TestDeleteBYOCKeyspaceRoleWithoutImportRoleNameIsNoop(t *testing.T) {
	ctx := context.Background()
	roleName := resourceName("test", "10000001")
	role := &iamv1beta1.Role{ObjectMeta: metav1.ObjectMeta{Name: roleName}}
	rolePolicy := &iamv1beta1.RolePolicy{ObjectMeta: metav1.ObjectMeta{Name: rolePolicyName(roleName)}}
	cli := newFakeClient(role, rolePolicy)
	svc := newTestBYOCServiceWithoutImportRole(cli, consts.ProviderTypeEKS)

	resp, err := svc.DeleteKeyspaceRole(ctx, &dmpbv1.DeleteKeyspaceRoleRequest{
		DataplaneId:  "dp-1",
		KeyspaceName: "10000001",
	})
	require.NoError(t, err)
	require.NotNil(t, resp)
	require.NoError(t, cli.Get(ctx, types.NamespacedName{Name: roleName}, &iamv1beta1.Role{}))
	require.NoError(t, cli.Get(ctx, types.NamespacedName{Name: rolePolicyName(roleName)}, &iamv1beta1.RolePolicy{}))
}

func TestGetStaticBYOCKeyspaceRoleDoesNotValidateGeneratedRoleName(t *testing.T) {
	ctx := context.Background()
	svc := newTestServiceWithConfig(newFakeClient(
		testDataPlane(),
		testBucket(v1alpha1.BucketPhaseReady),
	), config.RootConfig{
		Namespace:          "default",
		ResourceNamePrefix: strings.Repeat("p", 70),
		ControllerConfig: config.ControllerConfig{
			Provider: consts.ProviderTypeEKS,
		},
		BYOC: config.BYOCConfig{
			ImportRoleName: "tidbcloud-import-role",
		},
	})

	resp, err := svc.GetKeyspaceRole(ctx, &dmpbv1.GetKeyspaceRoleRequest{
		DataplaneId:   "dp-1",
		KeyspaceName:  "10000001",
		BucketId:      "bucket-1",
		AllowedPrefix: "10000001",
	})
	require.NoError(t, err)
	require.Equal(t, "arn:aws:iam::123456789012:role/tidbcloud-import-role", resp.RoleArn)
}

func TestGetStaticBYOCKeyspaceRoleRejectsMissingContract(t *testing.T) {
	for _, tc := range []struct {
		name          string
		bucketID      string
		allowedPrefix string
		message       string
	}{
		{
			name:          "missing bucket",
			allowedPrefix: "10000001",
			message:       "bucket_id is required for static import role",
		},
		{
			name:     "missing prefix",
			bucketID: "bucket-1",
			message:  "keyspace_role.allowed_prefix is required",
		},
	} {
		t.Run(tc.name, func(t *testing.T) {
			svc := newTestStaticImportRoleService(newFakeClient(), consts.ProviderTypeEKS)
			resp, err := svc.GetKeyspaceRole(context.Background(), &dmpbv1.GetKeyspaceRoleRequest{
				DataplaneId:   "dp-1",
				KeyspaceName:  "10000001",
				BucketId:      tc.bucketID,
				AllowedPrefix: tc.allowedPrefix,
			})
			require.Nil(t, resp)
			require.True(t, bizerr.Is(err, bizerr.ParameterInvalid))
			require.ErrorContains(t, err, tc.message)
		})
	}
}

func TestGetStaticBYOCKeyspaceRoleRejectsNotReadyBucket(t *testing.T) {
	ctx := context.Background()
	svc := newTestStaticImportRoleService(newFakeClient(
		testDataPlane(),
		testBucket(v1alpha1.BucketPhaseCreating),
	), consts.ProviderTypeEKS)

	resp, err := svc.GetKeyspaceRole(ctx, &dmpbv1.GetKeyspaceRoleRequest{
		DataplaneId:   "dp-1",
		KeyspaceName:  "10000001",
		BucketId:      "bucket-1",
		AllowedPrefix: "10000001",
	})
	require.Nil(t, resp)
	require.True(t, bizerr.Is(err, bizerr.ResourceNotReady))
	require.ErrorContains(t, err, "bucket is not ready")
}

func TestCreateKeyspaceRoleRejectsInvalidAllowedPrefix(t *testing.T) {
	for _, tc := range []struct {
		name          string
		allowedPrefix string
		message       string
	}{
		{
			name:          "empty",
			allowedPrefix: " / ",
			message:       "keyspace_role.allowed_prefix is required",
		},
		{
			name:          "star wildcard",
			allowedPrefix: "10000001/*",
			message:       "keyspace_role.allowed_prefix must not contain wildcard characters",
		},
		{
			name:          "question wildcard",
			allowedPrefix: "1000000?",
			message:       "keyspace_role.allowed_prefix must not contain wildcard characters",
		},
		{
			name:          "unsupported characters",
			allowedPrefix: "${aws:username}",
			message:       "keyspace_role.allowed_prefix contains unsupported characters",
		},
	} {
		t.Run(tc.name, func(t *testing.T) {
			svc := newTestService(newFakeClient())
			resp, err := svc.CreateKeyspaceRole(context.Background(), &dmpbv1.CreateKeyspaceRoleRequest{
				DataplaneId: "dp-1",
				KeyspaceRole: &dmpbv1.KeyspaceRole{
					KeyspaceName:  "10000001",
					BucketId:      "bucket-1",
					AllowedPrefix: tc.allowedPrefix,
					ExternalId:    "10000001",
					Tags:          map[string]string{consts.TenantLabelKey: "tenant-1"},
				},
			})
			require.Nil(t, resp)
			require.True(t, bizerr.Is(err, bizerr.ParameterInvalid))
			require.ErrorContains(t, err, tc.message)
		})
	}
}

func TestCreateKeyspaceRoleRejectsInvalidGeneratedRoleName(t *testing.T) {
	svc := newTestService(newFakeClient())
	resp, err := svc.CreateKeyspaceRole(context.Background(), &dmpbv1.CreateKeyspaceRoleRequest{
		DataplaneId: "dp-1",
		KeyspaceRole: &dmpbv1.KeyspaceRole{
			KeyspaceName:  strings.Repeat("a", 60),
			BucketId:      "bucket-1",
			AllowedPrefix: "10000001",
			ExternalId:    "10000001",
			Tags:          map[string]string{consts.TenantLabelKey: "tenant-1"},
		},
	})
	require.Nil(t, resp)
	require.True(t, bizerr.Is(err, bizerr.ParameterInvalid))
	require.ErrorContains(t, err, "generated keyspace role name must not exceed 64 characters")
}

func TestGetKeyspaceRoleReady(t *testing.T) {
	ctx := context.Background()
	roleName := resourceName("test", "10000001")
	role := &iamv1beta1.Role{
		ObjectMeta: metav1.ObjectMeta{
			Name: roleName,
			Labels: map[string]string{
				consts.DataplaneLabelKey: "dp-1",
				keyspaceNameLabelKey:     "10000001",
			},
			Annotations: map[string]string{
				keyspaceRoleBucketAnnotation: "bucket-1",
				keyspaceRolePrefixAnnotation: "10000001",
				keyspaceRoleExtIDAnnotation:  "10000001",
			},
		},
	}
	role.Status.AtProvider.Arn = ptr.To("arn:aws:iam::123456789012:role/test-keyspace-10000001")
	role.Status.SetConditions(xpv1.Available(), xpv1.ReconcileSuccess())
	rolePolicy := &iamv1beta1.RolePolicy{
		ObjectMeta: metav1.ObjectMeta{Name: rolePolicyName(roleName)},
	}
	rolePolicy.Status.SetConditions(xpv1.Available(), xpv1.ReconcileSuccess())

	svc := newTestService(newFakeClient(role, rolePolicy))
	resp, err := svc.GetKeyspaceRole(ctx, &dmpbv1.GetKeyspaceRoleRequest{
		DataplaneId:  "dp-1",
		KeyspaceName: "10000001",
	})
	require.NoError(t, err)
	require.Equal(t, dmpbv1.KeyspaceRole_Ready, resp.Status)
	require.Equal(t, "arn:aws:iam::123456789012:role/test-keyspace-10000001", resp.RoleArn)
	require.Equal(t, "bucket-1", resp.BucketId)
	require.Equal(t, "10000001", resp.AllowedPrefix)
}

func TestGetACKKeyspaceRoleReady(t *testing.T) {
	ctx := context.Background()
	roleName := resourceName("test", "10000001")
	role := &ramv1alpha1.Role{
		ObjectMeta: metav1.ObjectMeta{
			Name: roleName,
			Labels: map[string]string{
				consts.DataplaneLabelKey: "dp-1",
				keyspaceNameLabelKey:     "10000001",
			},
			Annotations: map[string]string{
				keyspaceRoleBucketAnnotation: "bucket-1",
				keyspaceRolePrefixAnnotation: "10000001",
				keyspaceRoleExtIDAnnotation:  "10000001",
			},
		},
	}
	role.Status.AtProvider.Arn = ptr.To("acs:ram::123456789012:role/test-keyspace-10000001")
	role.Status.SetConditions(xpv1.Available(), xpv1.ReconcileSuccess())
	policy := &ramv1alpha1.Policy{
		ObjectMeta: metav1.ObjectMeta{Name: rolePolicyName(roleName)},
	}
	policy.Status.SetConditions(xpv1.Available(), xpv1.ReconcileSuccess())
	attachment := &ramv1alpha1.RolePolicyAttachment{
		ObjectMeta: metav1.ObjectMeta{Name: rolePolicyAttachmentName(roleName)},
	}
	attachment.Status.SetConditions(xpv1.Available(), xpv1.ReconcileSuccess())

	svc := newTestServiceWithProvider(newFakeClient(role, policy, attachment), consts.ProviderTypeACK)
	resp, err := svc.GetKeyspaceRole(ctx, &dmpbv1.GetKeyspaceRoleRequest{
		DataplaneId:  "dp-1",
		KeyspaceName: "10000001",
	})
	require.NoError(t, err)
	require.Equal(t, dmpbv1.KeyspaceRole_Ready, resp.Status)
	require.Equal(t, "acs:ram::123456789012:role/test-keyspace-10000001", resp.RoleArn)
	require.Equal(t, "bucket-1", resp.BucketId)
	require.Equal(t, "10000001", resp.AllowedPrefix)
}

func TestCreateKeyspaceRoleRejectsNotReadyBucket(t *testing.T) {
	ctx := context.Background()
	svc := newTestService(newFakeClient(
		testDataPlane(),
		testAPPBox(),
		testBucket(v1alpha1.BucketPhaseCreating),
	))

	resp, err := svc.CreateKeyspaceRole(ctx, &dmpbv1.CreateKeyspaceRoleRequest{
		DataplaneId: "dp-1",
		KeyspaceRole: &dmpbv1.KeyspaceRole{
			KeyspaceName:  "10000001",
			BucketId:      "bucket-1",
			AllowedPrefix: "10000001",
			ExternalId:    "10000001",
			Tags:          map[string]string{consts.TenantLabelKey: "tenant-1"},
		},
	})
	require.Error(t, err)
	require.Nil(t, resp)
}

func TestDeleteKeyspaceRoleDeletesRolePolicyAndRole(t *testing.T) {
	ctx := context.Background()
	roleName := resourceName("test", "10000001")
	role := &iamv1beta1.Role{ObjectMeta: metav1.ObjectMeta{
		Name:   roleName,
		Labels: map[string]string{consts.DataplaneLabelKey: "dp-1"},
	}}
	rolePolicy := &iamv1beta1.RolePolicy{ObjectMeta: metav1.ObjectMeta{Name: rolePolicyName(roleName)}}
	cli := newFakeClient(role, rolePolicy)
	svc := newTestService(cli)

	resp, err := svc.DeleteKeyspaceRole(ctx, &dmpbv1.DeleteKeyspaceRoleRequest{
		DataplaneId:  "dp-1",
		KeyspaceName: "10000001",
	})
	require.NoError(t, err)
	require.NotNil(t, resp)

	err = cli.Get(ctx, types.NamespacedName{Name: rolePolicyName(roleName)}, &iamv1beta1.RolePolicy{})
	require.True(t, apierrors.IsNotFound(err))
	err = cli.Get(ctx, types.NamespacedName{Name: roleName}, &iamv1beta1.Role{})
	require.True(t, apierrors.IsNotFound(err))
}

func TestDeleteACKKeyspaceRoleDeletesAttachmentPolicyAndRole(t *testing.T) {
	ctx := context.Background()
	roleName := resourceName("test", "10000001")
	role := &ramv1alpha1.Role{ObjectMeta: metav1.ObjectMeta{
		Name:   roleName,
		Labels: map[string]string{consts.DataplaneLabelKey: "dp-1"},
	}}
	policy := &ramv1alpha1.Policy{ObjectMeta: metav1.ObjectMeta{Name: rolePolicyName(roleName)}}
	attachment := &ramv1alpha1.RolePolicyAttachment{ObjectMeta: metav1.ObjectMeta{Name: rolePolicyAttachmentName(roleName)}}
	cli := newFakeClient(role, policy, attachment)
	svc := newTestServiceWithProvider(cli, consts.ProviderTypeACK)

	resp, err := svc.DeleteKeyspaceRole(ctx, &dmpbv1.DeleteKeyspaceRoleRequest{
		DataplaneId:  "dp-1",
		KeyspaceName: "10000001",
	})
	require.NoError(t, err)
	require.NotNil(t, resp)

	err = cli.Get(ctx, types.NamespacedName{Name: rolePolicyAttachmentName(roleName)}, &ramv1alpha1.RolePolicyAttachment{})
	require.True(t, apierrors.IsNotFound(err))
	err = cli.Get(ctx, types.NamespacedName{Name: rolePolicyName(roleName)}, &ramv1alpha1.Policy{})
	require.True(t, apierrors.IsNotFound(err))
	err = cli.Get(ctx, types.NamespacedName{Name: roleName}, &ramv1alpha1.Role{})
	require.True(t, apierrors.IsNotFound(err))
}

func TestDeleteKeyspaceRoleIgnoresNotFound(t *testing.T) {
	resp, err := newTestService(newFakeClient()).DeleteKeyspaceRole(context.Background(), &dmpbv1.DeleteKeyspaceRoleRequest{
		DataplaneId:  "dp-1",
		KeyspaceName: "10000001",
	})
	require.NoError(t, err)
	require.NotNil(t, resp)
}

func TestGetKeyspaceRoleRejectsMismatchedDataplaneOwner(t *testing.T) {
	ctx := context.Background()
	roleName := resourceName("test", "10000001")
	role := &iamv1beta1.Role{
		ObjectMeta: metav1.ObjectMeta{
			Name: roleName,
			Labels: map[string]string{
				consts.DataplaneLabelKey: "dp-2",
				keyspaceNameLabelKey:     "10000001",
			},
		},
	}
	svc := newTestService(newFakeClient(role))

	resp, err := svc.GetKeyspaceRole(ctx, &dmpbv1.GetKeyspaceRoleRequest{
		DataplaneId:  "dp-1",
		KeyspaceName: "10000001",
	})
	require.Nil(t, resp)
	require.True(t, bizerr.Is(err, bizerr.ResourceNotFound))
}

func TestDeleteKeyspaceRoleRejectsMismatchedDataplaneOwner(t *testing.T) {
	ctx := context.Background()
	roleName := resourceName("test", "10000001")
	role := &iamv1beta1.Role{
		ObjectMeta: metav1.ObjectMeta{
			Name:   roleName,
			Labels: map[string]string{consts.DataplaneLabelKey: "dp-2"},
		},
	}
	rolePolicy := &iamv1beta1.RolePolicy{ObjectMeta: metav1.ObjectMeta{Name: rolePolicyName(roleName)}}
	cli := newFakeClient(role, rolePolicy)
	svc := newTestService(cli)

	resp, err := svc.DeleteKeyspaceRole(ctx, &dmpbv1.DeleteKeyspaceRoleRequest{
		DataplaneId:  "dp-1",
		KeyspaceName: "10000001",
	})
	require.Nil(t, resp)
	require.True(t, bizerr.Is(err, bizerr.ResourceNotFound))
	require.NoError(t, cli.Get(ctx, types.NamespacedName{Name: rolePolicyName(roleName)}, &iamv1beta1.RolePolicy{}))
}

func TestCreateKeyspaceRoleRejectsForeignOwnedAWSResource(t *testing.T) {
	ctx := context.Background()
	roleName := resourceName("test", "10000001")
	role := &iamv1beta1.Role{ObjectMeta: metav1.ObjectMeta{
		Name:   roleName,
		Labels: map[string]string{consts.DataplaneLabelKey: "dp-2"},
	}}
	cli := newFakeClient(role, testDataPlane(), testAPPBox(), testBucket(v1alpha1.BucketPhaseReady))
	svc := newTestService(cli)

	resp, err := svc.CreateKeyspaceRole(ctx, &dmpbv1.CreateKeyspaceRoleRequest{
		DataplaneId: "dp-1",
		KeyspaceRole: &dmpbv1.KeyspaceRole{
			KeyspaceName:  "10000001",
			BucketId:      "bucket-1",
			AllowedPrefix: "10000001",
			ExternalId:    "10000001",
			Tags:          map[string]string{consts.TenantLabelKey: "tenant-1"},
		},
	})
	require.Nil(t, resp)
	require.True(t, bizerr.Is(err, bizerr.ResourceConflict))
}

func TestCreateACKKeyspaceRoleRejectsForeignOwnedResource(t *testing.T) {
	ctx := context.Background()
	roleName := resourceName("test", "10000001")
	role := &ramv1alpha1.Role{ObjectMeta: metav1.ObjectMeta{
		Name:   roleName,
		Labels: map[string]string{consts.DataplaneLabelKey: "dp-2"},
	}}
	cli := newFakeClient(role, testDataPlane(), testAPPBoxWithProvider(consts.ProviderTypeACK), testBucket(v1alpha1.BucketPhaseReady))
	svc := newTestServiceWithProvider(cli, consts.ProviderTypeACK)

	resp, err := svc.CreateKeyspaceRole(ctx, &dmpbv1.CreateKeyspaceRoleRequest{
		DataplaneId: "dp-1",
		KeyspaceRole: &dmpbv1.KeyspaceRole{
			KeyspaceName:  "10000001",
			BucketId:      "bucket-1",
			AllowedPrefix: "10000001",
			ExternalId:    "10000001",
			Tags:          map[string]string{consts.TenantLabelKey: "tenant-1"},
		},
	})
	require.Nil(t, resp)
	require.True(t, bizerr.Is(err, bizerr.ResourceConflict))
}

func TestBuildS3PolicyKeepsMultipartListDenied(t *testing.T) {
	policy, err := buildS3Policy("shared-bucket", "/10000001/")
	require.NoError(t, err)
	require.NotContains(t, policy, "s3:ListBucketMultipartUploads")
	require.Contains(t, policy, "s3:ListMultipartUploadParts")
	require.Contains(t, policy, "arn:aws:s3:::shared-bucket/10000001/*")
	requirePolicyListPrefixes(t, policy, "s3:prefix", []string{"10000001/*"})
}

func TestBuildOSSPolicyKeepsMultipartListDenied(t *testing.T) {
	policy, err := buildOSSPolicy("shared-bucket", "/10000001/")
	require.NoError(t, err)
	require.NotContains(t, policy, "oss:ListMultipartUploads")
	require.Contains(t, policy, "oss:ListParts")
	require.Contains(t, policy, "10000001")
	require.Contains(t, policy, "acs:oss:*:*:shared-bucket/10000001/*")
	requirePolicyListPrefixes(t, policy, "oss:Prefix", []string{"10000001/*"})
}

func TestResourceNameUsesKeyspaceNameWithoutHash(t *testing.T) {
	name := resourceName(strings.Repeat("p", 70), "10000001")
	require.Equal(t, strings.Repeat("p", 70)+"-keyspace-10000001", name)
	require.Equal(t, "tidbcloud-keyspace-10000001", resourceName("", "10000001"))
}

func TestRolePolicyNamesAppendSuffix(t *testing.T) {
	roleName := strings.Repeat("r", 64)
	require.Equal(t, roleName+"-policy", rolePolicyName(roleName))
	require.Equal(t, roleName+"-policy-attachment", rolePolicyAttachmentName(roleName))
}

func TestValidateRoleName(t *testing.T) {
	require.NoError(t, validateRoleName("tidbcloud-keyspace-10000001"))

	err := validateRoleName(resourceName(strings.Repeat("p", 70), "10000001"))
	require.True(t, bizerr.Is(err, bizerr.ParameterInvalid))
	require.ErrorContains(t, err, "generated keyspace role name must not exceed 64 characters")

	err = validateRoleName(resourceName("test", "Keyspace_1"))
	require.True(t, bizerr.Is(err, bizerr.ParameterInvalid))
	require.ErrorContains(t, err, "generated keyspace role name is invalid")
}

func TestBuildTrustPolicyRejectsMissingFields(t *testing.T) {
	policy, err := buildTrustPolicy("", "arn:aws:iam::123456789012:role/node-role", "10000001")
	require.Empty(t, policy)
	require.True(t, bizerr.Is(err, bizerr.ParameterInvalid))
}

func TestBuildRAMTrustPolicyLowercasesPrincipalARN(t *testing.T) {
	policy, err := buildRAMTrustPolicy("123456789012", "acs:ram::123456789012:role/NodeRole", "10000001")
	require.NoError(t, err)

	statement := decodePolicy(t, policy)["Statement"].([]any)[0].(map[string]any)
	condition := statement["Condition"].(map[string]any)
	require.Equal(t, "acs:ram::123456789012:role/noderole", condition["StringEquals"].(map[string]any)["acs:PrincipalARN"])
}

func newTestService(cli client.Client) dmpbv1.KeyspaceRoleServiceServer {
	return newTestServiceWithProvider(cli, consts.ProviderTypeEKS)
}

func newTestServiceWithProvider(cli client.Client, provider string) dmpbv1.KeyspaceRoleServiceServer {
	return newTestServiceWithConfig(cli, config.RootConfig{
		Namespace:          "default",
		ResourceNamePrefix: "test",
		ControllerConfig: config.ControllerConfig{
			Provider: provider,
		},
	})
}

func newTestStaticImportRoleService(cli client.Client, provider string) dmpbv1.KeyspaceRoleServiceServer {
	return newTestServiceWithConfig(cli, config.RootConfig{
		Namespace:          "default",
		ResourceNamePrefix: "test",
		ControllerConfig: config.ControllerConfig{
			Provider: provider,
		},
		BYOC: config.BYOCConfig{
			BYOCPrivateLink: config.BYOCPrivateLinkConfig{
				ControlPlaneVPCID: "vpc-1",
			},
			ImportRoleName: "tidbcloud-import-role",
		},
	})
}

func newTestBYOCServiceWithoutImportRole(cli client.Client, provider string) dmpbv1.KeyspaceRoleServiceServer {
	return newTestServiceWithConfig(cli, config.RootConfig{
		Namespace:          "default",
		ResourceNamePrefix: "test",
		ControllerConfig: config.ControllerConfig{
			Provider: provider,
		},
		BYOC: config.BYOCConfig{
			BYOCPrivateLink: config.BYOCPrivateLinkConfig{
				ControlPlaneVPCID: "vpc-1",
			},
		},
	})
}

func newTestServiceWithConfig(cli client.Client, cfg config.RootConfig) dmpbv1.KeyspaceRoleServiceServer {
	return NewService(
		cli,
		repository.NewDataPlaneReader(cli),
		repository.NewAPPBoxReader(cli),
		repository.NewBucketReader(cli),
		cfg,
	)
}

func newFakeClient(objs ...client.Object) client.Client {
	return fake.NewClientBuilder().
		WithScheme(dpmscheme.Scheme).
		WithObjects(objs...).
		Build()
}

func testDataPlane() *v1alpha1.DataPlane {
	return &v1alpha1.DataPlane{
		ObjectMeta: metav1.ObjectMeta{
			Namespace: "default",
			Name:      "dp-1",
		},
		Spec: v1alpha1.DataPlaneSpec{
			AccountRef: v1alpha1.AccountReference{
				AccountID:          "123456789012",
				ProviderConfigName: "provider-dp",
			},
		},
		Status: v1alpha1.DataPlaneStatus{
			AppBox: &v1alpha1.Reference{Name: "appbox-1"},
		},
	}
}

func testAPPBox() *v1alpha1.APPBox {
	return testAPPBoxWithProvider(consts.ProviderTypeEKS)
}

func testAPPBoxWithProvider(provider string) *v1alpha1.APPBox {
	iamStatus := v1alpha1.IAMStatus{
		NodeRoleARN: "arn:aws:iam::123456789012:role/node-role",
	}
	if provider == consts.ProviderTypeACK {
		iamStatus.NodeRoleARN = "acs:ram::123456789012:role/NodeRole"
		iamStatus.NodeRoleName = "NodeRole"
	}
	return &v1alpha1.APPBox{
		ObjectMeta: metav1.ObjectMeta{
			Namespace: "default",
			Name:      "appbox-1",
		},
		Spec: v1alpha1.APPBoxSpec{
			Provider: provider,
		},
		Status: v1alpha1.APPBoxStatus{
			IAM: iamStatus,
		},
	}
}

func testBucket(phase v1alpha1.BucketPhase) *v1alpha1.Bucket {
	return &v1alpha1.Bucket{
		ObjectMeta: metav1.ObjectMeta{
			Namespace: "default",
			Name:      "dp-1-bucket-1",
		},
		Spec: v1alpha1.BucketSpec{
			DataplaneId: "dp-1",
			Id:          "bucket-1",
		},
		Status: v1alpha1.BucketStatus{
			Phase:  phase,
			Bucket: &v1alpha1.Reference{Name: "shared-bucket"},
		},
	}
}

func decodePolicy(t *testing.T, policy string) map[string]any {
	t.Helper()
	var doc map[string]any
	require.NoError(t, json.Unmarshal([]byte(policy), &doc))
	return doc
}

func requirePolicyListPrefixes(t *testing.T, policy, conditionKey string, expected []string) {
	t.Helper()
	statements := decodePolicy(t, policy)["Statement"].([]any)
	for _, item := range statements {
		statement := item.(map[string]any)
		condition, ok := statement["Condition"].(map[string]any)
		if !ok {
			continue
		}
		stringLike, ok := condition["StringLike"].(map[string]any)
		if !ok {
			continue
		}
		value, ok := stringLike[conditionKey]
		if !ok {
			continue
		}
		require.Equal(t, expected, stringSlice(t, value))
		return
	}
	require.Failf(t, "missing list prefix condition", "condition key %s not found", conditionKey)
}

func stringSlice(t *testing.T, value any) []string {
	t.Helper()
	switch typed := value.(type) {
	case string:
		return []string{typed}
	case []any:
		out := make([]string, 0, len(typed))
		for _, item := range typed {
			out = append(out, item.(string))
		}
		return out
	default:
		require.Failf(t, "unexpected condition value type", "%T", value)
		return nil
	}
}