package resourcepool

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

	"github.com/google/uuid"
	"github.com/stretchr/testify/require"
	"github.com/tidbcloud/tiinf/sdk/auditlog"
	"go.uber.org/mock/gomock"
	"google.golang.org/genproto/googleapis/rpc/errdetails"
	"google.golang.org/grpc"
	"google.golang.org/grpc/codes"
	"google.golang.org/grpc/status"
	"google.golang.org/protobuf/proto"
	"google.golang.org/protobuf/types/known/fieldmaskpb"

	globalcommonv1 "github.com/tidbcloud/tidb-management-service/api/spec/global/common/v1"
	globalngv1 "github.com/tidbcloud/tidb-management-service/api/spec/global/nextgen/v1"
	openapicommonv1beta1 "github.com/tidbcloud/tidb-management-service/api/spec/tidb_cloud_open_api/common/v1beta1"
	openapingv1beta2 "github.com/tidbcloud/tidb-management-service/api/spec/tidb_cloud_open_api/nextgen/v1beta2"
	"github.com/tidbcloud/tidb-management-service/tidb-mgmt-service/domain/adapter/eventbus"
	"github.com/tidbcloud/tidb-management-service/tidb-mgmt-service/domain/adapter/nextgenglobal"
	"github.com/tidbcloud/tidb-management-service/tidb-mgmt-service/domain/adapter/nextgenglobal/mock_nextgenglobal"
	"github.com/tidbcloud/tidb-management-service/utilities/grpc/metadata"
)

func TestServerCreateResourcePoolConvertsTrustedOrganizationAndCreator(t *testing.T) {
	t.Parallel()

	ctrl := gomock.NewController(t)
	nextgenCli := mock_nextgenglobal.NewMockNextGenGlobal(ctrl)
	auditLogSender := &recordingAuditLogSender{}
	server := NewServer(nextgenCli, auditLogSender)
	ctx := checkedContext(t)

	nextgenCli.EXPECT().CreateResourcePool(gomock.Any(), gomock.Any()).DoAndReturn(
		func(_ context.Context, req *globalngv1.CreateResourcePoolRequest, _ ...grpc.CallOption) (*globalngv1.ResourcePoolProfile, error) {
			require.Equal(t, uint64(1001), req.GetResourcePool().GetOwnerId())
			require.Equal(t, globalngv1.ResourcePool_ORGANIZATION, req.GetResourcePool().GetOwnerType())
			require.Equal(t, globalcommonv1.ServicePlan_BYOC, req.GetResourcePool().GetServicePlan())
			require.Equal(t, globalcommonv1.CloudProvider_AWS, req.GetResourcePool().GetProviderId())
			require.Equal(t, globalcommonv1.Region_AWS_US_WEST_2, req.GetResourcePool().GetRegionId())
			require.Equal(t, globalngv1.ResourcePool_ZONAL, req.GetResourcePool().GetHighAvailabilityType())
			require.Equal(t, "pool-a", req.GetResourcePool().GetName())
			require.Equal(t, "10.0.0.0/16", req.GetResourcePool().GetCidr())
			require.Equal(t, uint64(120), req.GetResourcePool().GetVcpusLimit())
			require.Equal(t, globalngv1.ResourcePool_LIMITED, req.GetVcpuLimitMode())
			require.Equal(t, "creator@example.com", req.GetCreatedBy())
			require.Equal(t, int64(8), req.GetBootstrapCapacity().GetTikv().GetNodeSpec().GetCpu())
			require.Equal(t, int64(3), req.GetBootstrapCapacity().GetTikv().GetNodeCount())
			require.Equal(t, []*globalngv1.Storage{{Name: "data", SizeGi: 500}},
				req.GetBootstrapCapacity().GetTikv().GetStorages())
			require.Nil(t, req.GetDemand())
			require.False(t, req.GetValidateOnly())

			pool, ok := proto.Clone(req.GetResourcePool()).(*globalngv1.ResourcePool)
			require.True(t, ok)
			pool.Id = 42
			pool.LifecycleState = globalngv1.ResourcePool_CREATING
			pool.InstanceCount = 0
			return &globalngv1.ResourcePoolProfile{
				ResourcePool: pool,
				Region: &globalcommonv1.Region{
					DisplayName: "Oregon (us-west-2)",
				},
				Capacity: &globalngv1.ResourcePoolCapacity{
					Tikv: req.GetBootstrapCapacity().GetTikv(),
				},
			}, nil
		},
	)

	response, err := server.CreateResourcePool(ctx, &openapingv1beta2.CreateResourcePoolRequest{
		ResourcePool: &openapingv1beta2.ResourcePool{
			DisplayName:          "pool-a",
			RegionId:             "aws-us-west-2",
			VpcCidr:              "10.0.0.0/16",
			HighAvailabilityType: openapingv1beta2.ResourcePool_ZONAL,
			VcpuLimit: &openapingv1beta2.VcpuLimit{
				Policy: &openapingv1beta2.VcpuLimit_LimitedVcpus{LimitedVcpus: 120},
			},
			Capacity: &openapingv1beta2.ResourcePoolCapacity{
				Tikv: &openapingv1beta2.ComponentCapacity{
					NodeSpec:  &openapingv1beta2.NodeSpecSelection{Cpu: 8, MemoryGi: 32},
					NodeCount: proto.Int64(3),
					Storages: []*openapingv1beta2.Storage{
						{Name: "data", SizeGi: 500},
					},
				},
			},
			CloudResourceTags: map[string]string{"environment": "production-secret"},
		},
		ValidateOnly: false,
	})
	require.NoError(t, err)
	require.Equal(t, "resourcePools/42", response.GetName())
	require.Equal(t, "42", response.GetResourcePoolId())
	require.Equal(t, "aws-us-west-2", response.GetRegionId())
	require.Equal(t, "Oregon (us-west-2)", response.GetRegionDisplayName())
	require.Equal(t, openapicommonv1beta1.Region_aws, response.GetCloudProvider())
	require.Equal(t, openapingv1beta2.ResourcePool_CREATING, response.GetState())
	require.Nil(t, response.DataplaneId)
	require.Nil(t, response.ZoneConfig)
	require.Len(t, auditLogSender.records, 1)
	record := auditLogSender.records[0]
	require.Equal(t, auditlog.EventCreateResourcePool, record.eventType)
	require.Equal(t, uint64(1001), record.reqMeta.OrgID)
	require.NoError(t, record.err)
	require.Equal(t, "42", record.details[auditlog.KeyResourcePoolId])
	require.Equal(t, "pool-a", record.details[auditlog.KeyDisplayName])
	require.Equal(t, "creator@example.com", record.details[auditlog.KeyCreatedBy])
	require.Equal(t, openapicommonv1beta1.ServicePlan_BYOC, record.details[auditlog.KeyServicePlan])
	require.Equal(t, []string{"environment"}, record.details[auditlog.KeyChangedTagKeys])
	detailJSON, err := json.Marshal(record.details)
	require.NoError(t, err)
	require.NotContains(t, string(detailJSON), "production-secret")
}

func TestServerResumeResourcePoolUsesGeneratedInternalOperationID(t *testing.T) {
	t.Parallel()

	ctrl := gomock.NewController(t)
	nextgenCli := mock_nextgenglobal.NewMockNextGenGlobal(ctrl)
	auditLogSender := &recordingAuditLogSender{}
	server := NewServer(nextgenCli, auditLogSender)

	nextgenCli.EXPECT().ResumeResourcePool(gomock.Any(), gomock.Any()).DoAndReturn(
		func(
			_ context.Context,
			req *globalngv1.ResumeResourcePoolRequest,
			_ ...grpc.CallOption,
		) (*globalngv1.ResumeResourcePoolResponse, error) {
			require.Equal(t, uint64(42), req.GetResourcePoolId())
			_, err := uuid.Parse(req.GetOperationId())
			require.NoError(t, err)
			return &globalngv1.ResumeResourcePoolResponse{OperationId: req.GetOperationId()}, nil
		},
	)

	response, err := server.ResumeResourcePool(checkedContext(t), &openapingv1beta2.ResumeResourcePoolRequest{
		ResourcePoolId: "42",
	})
	require.NoError(t, err)
	require.NotNil(t, response)
	require.Len(t, auditLogSender.records, 1)
	require.Equal(t, auditlog.EventUpdateResourcePool, auditLogSender.records[0].eventType)
	require.Equal(t, resourcePoolResumeAction, auditLogSender.records[0].details[auditlog.KeyAction])
	require.Equal(t, "42", auditLogSender.records[0].details[auditlog.KeyResourcePoolId])
	require.NoError(t, auditLogSender.records[0].err)
}

func TestServerResumeResourcePoolPropagatesGlobalAdmissionError(t *testing.T) {
	t.Parallel()

	ctrl := gomock.NewController(t)
	nextgenCli := mock_nextgenglobal.NewMockNextGenGlobal(ctrl)
	auditLogSender := &recordingAuditLogSender{}
	server := NewServer(nextgenCli, auditLogSender)
	wantErr := status.Error(codes.FailedPrecondition, "resource pool is not paused")
	nextgenCli.EXPECT().ResumeResourcePool(gomock.Any(), gomock.Any()).Return(nil, wantErr)
	nextgenCli.EXPECT().GetResourcePool(gomock.Any(), &globalngv1.GetResourcePoolRequest{
		ResourcePoolId: 42,
	}).Return(&globalngv1.ResourcePoolProfile{ResourcePool: &globalngv1.ResourcePool{
		Id:             42,
		LifecycleState: globalngv1.ResourcePool_PAUSED,
	}}, nil)

	response, err := server.ResumeResourcePool(checkedContext(t), &openapingv1beta2.ResumeResourcePoolRequest{
		ResourcePoolId: "42",
	})
	require.Nil(t, response)
	require.ErrorIs(t, err, wantErr)
	require.Len(t, auditLogSender.records, 1)
	require.ErrorIs(t, auditLogSender.records[0].err, wantErr)
}

func TestServerResumeResourcePoolTreatsConcurrentRetryAsSuccess(t *testing.T) {
	t.Parallel()

	ctrl := gomock.NewController(t)
	nextgenCli := mock_nextgenglobal.NewMockNextGenGlobal(ctrl)
	auditLogSender := &recordingAuditLogSender{}
	server := NewServer(nextgenCli, auditLogSender)

	nextgenCli.EXPECT().ResumeResourcePool(gomock.Any(), gomock.Any()).Return(
		nil,
		status.Error(codes.FailedPrecondition, "resource pool is already resuming"),
	)
	nextgenCli.EXPECT().GetResourcePool(gomock.Any(), &globalngv1.GetResourcePoolRequest{
		ResourcePoolId: 42,
	}).Return(&globalngv1.ResourcePoolProfile{ResourcePool: &globalngv1.ResourcePool{
		Id:             42,
		LifecycleState: globalngv1.ResourcePool_RESUMING,
	}}, nil)

	response, err := server.ResumeResourcePool(checkedContext(t), &openapingv1beta2.ResumeResourcePoolRequest{
		ResourcePoolId: "42",
	})
	require.NoError(t, err)
	require.NotNil(t, response)
	require.Len(t, auditLogSender.records, 1)
	require.NoError(t, auditLogSender.records[0].err)
}

func TestNewResourcePoolAuditDetailsUsesProvidedServicePlan(t *testing.T) {
	t.Parallel()

	details := newResourcePoolAuditDetails(
		"42",
		&openapingv1beta2.ResourcePool{},
		globalcommonv1.ServicePlan_Premium_Reserved,
	)

	require.Equal(t, openapicommonv1beta1.ServicePlan_Premium_Reserved, details[auditlog.KeyServicePlan])
}

func TestServerCreateResourcePoolRebuildsPublicCIDRValidationDetails(t *testing.T) {
	t.Parallel()

	internalStatus, err := status.New(codes.InvalidArgument, "invalid CIDR").WithDetails(
		&globalngv1.ResourcePoolCidrValidationError{BlockingFindings: []*globalngv1.ResourcePoolCidrValidationFinding{{
			Code:    globalngv1.ResourcePoolCidrValidationFinding_INVALID_CIDR,
			Message: "The CIDR must be a canonical IPv4 network.",
		}}},
	)
	require.NoError(t, err)
	preserved := nextgenglobal.NewCIDRValidationStatusError(internalStatus.Err())
	require.NotNil(t, preserved)

	ctrl := gomock.NewController(t)
	nextgenCli := mock_nextgenglobal.NewMockNextGenGlobal(ctrl)
	auditLogSender := &recordingAuditLogSender{}
	server := NewServer(nextgenCli, auditLogSender)
	nextgenCli.EXPECT().CreateResourcePool(gomock.Any(), gomock.Any()).Return(nil, preserved)

	_, err = server.CreateResourcePool(checkedContext(t), &openapingv1beta2.CreateResourcePoolRequest{
		ResourcePool: &openapingv1beta2.ResourcePool{
			DisplayName: "pool-a",
			RegionId:    "aws-us-west-2",
			VpcCidr:     "10.0.1.1/16",
			VcpuLimit: &openapingv1beta2.VcpuLimit{
				Policy: &openapingv1beta2.VcpuLimit_Unlimited{Unlimited: true},
			},
			Capacity: &openapingv1beta2.ResourcePoolCapacity{Tikv: &openapingv1beta2.ComponentCapacity{
				NodeSpec:  &openapingv1beta2.NodeSpecSelection{Cpu: 8, MemoryGi: 32},
				NodeCount: proto.Int64(3),
				Storages:  []*openapingv1beta2.Storage{{Name: "data", SizeGi: 500}},
			}},
		},
	})

	require.Error(t, err)
	publicStatus := status.Convert(err)
	require.Equal(t, codes.InvalidArgument, publicStatus.Code())
	require.Len(t, publicStatus.Details(), 3)
	errorInfo, ok := publicStatus.Details()[0].(*errdetails.ErrorInfo)
	require.True(t, ok)
	require.Equal(t, "INVALID_CIDR", errorInfo.GetReason())
	badRequest, ok := publicStatus.Details()[1].(*errdetails.BadRequest)
	require.True(t, ok)
	require.Equal(t, "resourcePool.vpcCidr", badRequest.GetFieldViolations()[0].GetField())
	publicDetail, ok := publicStatus.Details()[2].(*openapingv1beta2.ResourcePoolCidrValidationError)
	require.True(t, ok)
	require.Equal(t, openapingv1beta2.ResourcePoolCidrValidationFinding_INVALID_CIDR,
		publicDetail.GetBlockingFindings()[0].GetCode())
	require.Len(t, auditLogSender.records, 1)
	require.Equal(t, auditlog.EventCreateResourcePool, auditLogSender.records[0].eventType)
	require.Error(t, auditLogSender.records[0].err)
}

func TestServerListResourcePoolsUsesSummaryViewAndOrganizationFilter(t *testing.T) {
	t.Parallel()

	ctrl := gomock.NewController(t)
	nextgenCli := mock_nextgenglobal.NewMockNextGenGlobal(ctrl)
	auditLogSender := &recordingAuditLogSender{}
	server := NewServer(nextgenCli, auditLogSender)
	ctx := checkedContext(t)

	nextgenCli.EXPECT().ListResourcePools(gomock.Any(), gomock.Any()).DoAndReturn(
		func(_ context.Context, req *globalngv1.ListResourcePoolsRequest, _ ...grpc.CallOption) (*globalngv1.ListResourcePoolsResponse, error) {
			require.Equal(t, []uint64{1001}, req.GetOwnerIds())
			require.Equal(t, globalngv1.ResourcePool_ORGANIZATION, req.GetOwnerType())
			require.Equal(t, globalcommonv1.ServicePlan_BYOC, req.GetServicePlan())
			require.Equal(t, globalngv1.ListResourcePoolsRequest_BASIC, req.GetView())
			require.Equal(t, uint32(10), req.GetPageSize())
			return &globalngv1.ListResourcePoolsResponse{
				ResourcePools: []*globalngv1.ResourcePoolProfile{{
					ResourcePool: &globalngv1.ResourcePool{
						Id:                   42,
						Name:                 "pool-a",
						ProviderId:           globalcommonv1.CloudProvider_AWS,
						RegionId:             globalcommonv1.Region_AWS_US_WEST_2,
						OwnerType:            globalngv1.ResourcePool_ORGANIZATION,
						OwnerId:              1001,
						ServicePlan:          globalcommonv1.ServicePlan_BYOC,
						LifecycleState:       globalngv1.ResourcePool_ACTIVE,
						HighAvailabilityType: globalngv1.ResourcePool_ZONAL,
						VcpusLimit:           120,
						VcpuLimitMode:        globalngv1.ResourcePool_LIMITED,
						InstanceCount:        3,
						DataplaneId:          "must-not-leak-into-summary",
					},
					Region: &globalcommonv1.Region{DisplayName: "Oregon (us-west-2)"},
				}},
				TotalSize:     1,
				NextPageToken: "next",
			}, nil
		},
	)

	response, err := server.ListResourcePools(ctx, &openapingv1beta2.ListResourcePoolsRequest{
		CloudProvider: openapicommonv1beta1.Region_aws,
		RegionIds:     []string{"aws-us-west-2"},
		PageSize:      10,
	})
	require.NoError(t, err)
	require.Len(t, response.GetResourcePools(), 1)
	require.Equal(t, "resourcePools/42", response.GetResourcePools()[0].GetName())
	require.Equal(t, int64(3), response.GetResourcePools()[0].GetTidbInstanceCount())
	require.Equal(t, openapingv1beta2.ResourcePool_ZONAL,
		response.GetResourcePools()[0].GetHighAvailabilityType())
	require.Equal(t, "Oregon (us-west-2)", response.GetResourcePools()[0].GetRegionDisplayName())
	require.Equal(t, int64(120), response.GetResourcePools()[0].GetVcpuLimit().GetLimitedVcpus())
	require.Equal(t, int32(1), response.GetTotalSize())
	require.Equal(t, "next", response.GetNextPageToken())
	require.Empty(t, auditLogSender.records)
}

func TestServerCreateResourcePoolValidateOnlyDoesNotWriteAuditLog(t *testing.T) {
	t.Parallel()

	ctrl := gomock.NewController(t)
	nextgenCli := mock_nextgenglobal.NewMockNextGenGlobal(ctrl)
	auditLogSender := &recordingAuditLogSender{}
	server := NewServer(nextgenCli, auditLogSender)
	nextgenCli.EXPECT().CreateResourcePool(gomock.Any(), gomock.Any()).DoAndReturn(
		func(_ context.Context, req *globalngv1.CreateResourcePoolRequest, _ ...grpc.CallOption) (*globalngv1.ResourcePoolProfile, error) {
			require.True(t, req.GetValidateOnly())
			return &globalngv1.ResourcePoolProfile{ResourcePool: req.GetResourcePool()}, nil
		},
	)

	_, err := server.CreateResourcePool(checkedContext(t), &openapingv1beta2.CreateResourcePoolRequest{
		ResourcePool: &openapingv1beta2.ResourcePool{
			DisplayName:          "pool-a",
			RegionId:             "aws-us-west-2",
			VpcCidr:              "10.0.0.0/16",
			HighAvailabilityType: openapingv1beta2.ResourcePool_ZONAL,
			VcpuLimit: &openapingv1beta2.VcpuLimit{
				Policy: &openapingv1beta2.VcpuLimit_Unlimited{Unlimited: true},
			},
			Capacity: testOpenAPIResourcePoolCapacity(3),
		},
		ValidateOnly: true,
	})

	require.NoError(t, err)
	require.Empty(t, auditLogSender.records)
}

func TestServerUpdateResourcePoolWritesOldAndNewNameToAuditLog(t *testing.T) {
	t.Parallel()

	ctrl := gomock.NewController(t)
	nextgenCli := mock_nextgenglobal.NewMockNextGenGlobal(ctrl)
	auditLogSender := &recordingAuditLogSender{}
	server := NewServer(nextgenCli, auditLogSender)
	oldProfile := testGlobalResourcePoolProfile("old-pool", map[string]string{"environment": "prod"})
	newProfile := testGlobalResourcePoolProfile("new-pool", map[string]string{"environment": "prod"})

	gomock.InOrder(
		nextgenCli.EXPECT().GetResourcePool(gomock.Any(), gomock.Any()).Return(oldProfile, nil),
		nextgenCli.EXPECT().UpdateResourcePoolName(gomock.Any(), gomock.Any()).DoAndReturn(
			func(_ context.Context, req *globalngv1.UpdateResourcePoolNameRequest, _ ...grpc.CallOption) (*globalngv1.ResourcePool, error) {
				require.Equal(t, uint64(42), req.GetResourcePoolId())
				require.Equal(t, "new-pool", req.GetName())
				return newProfile.GetResourcePool(), nil
			},
		),
		nextgenCli.EXPECT().GetResourcePool(gomock.Any(), gomock.Any()).Return(newProfile, nil),
	)

	response, err := server.UpdateResourcePool(checkedContext(t), &openapingv1beta2.UpdateResourcePoolRequest{
		ResourcePool: &openapingv1beta2.ResourcePool{
			ResourcePoolId: "42",
			DisplayName:    "new-pool",
		},
		UpdateMask: &fieldmaskpb.FieldMask{Paths: []string{"display_name"}},
	})

	require.NoError(t, err)
	require.Equal(t, "new-pool", response.GetDisplayName())
	require.Len(t, auditLogSender.records, 1)
	record := auditLogSender.records[0]
	require.Equal(t, auditlog.EventUpdateResourcePool, record.eventType)
	require.NoError(t, record.err)
	require.Equal(t, resourcePoolUpdateDisplayNameAction, record.details[auditlog.KeyAction])
	require.Equal(t, "old-pool", record.details[auditlog.KeyOldValue])
	require.Equal(t, "new-pool", record.details[auditlog.KeyNewValue])
}

func TestServerDeleteResourcePoolWritesFailedAuditLog(t *testing.T) {
	t.Parallel()

	ctrl := gomock.NewController(t)
	nextgenCli := mock_nextgenglobal.NewMockNextGenGlobal(ctrl)
	auditLogSender := &recordingAuditLogSender{}
	server := NewServer(nextgenCli, auditLogSender)
	deleteErr := errors.New("delete rejected")

	nextgenCli.EXPECT().GetResourcePool(gomock.Any(), gomock.Any()).Return(
		testGlobalResourcePoolProfile("pool-a", nil), nil,
	)
	nextgenCli.EXPECT().DeleteResourcePool(gomock.Any(), gomock.Any()).Return(nil, deleteErr)

	_, err := server.DeleteResourcePool(checkedContext(t), &openapingv1beta2.DeleteResourcePoolRequest{
		ResourcePoolId: "42",
	})

	require.ErrorIs(t, err, deleteErr)
	require.Len(t, auditLogSender.records, 1)
	record := auditLogSender.records[0]
	require.Equal(t, auditlog.EventDeleteResourcePool, record.eventType)
	require.ErrorIs(t, record.err, deleteErr)
	require.Equal(t, "42", record.details[auditlog.KeyResourcePoolId])
	require.Equal(t, "pool-a", record.details[auditlog.KeyDisplayName])
}

func TestServerResizeResourcePoolCapacityAuditLog(t *testing.T) {
	t.Parallel()

	for _, tt := range []struct {
		name          string
		validateOnly  bool
		wantAuditLogs int
	}{
		{name: "mutation", wantAuditLogs: 1},
		{name: "validate only", validateOnly: true},
	} {
		t.Run(tt.name, func(t *testing.T) {
			ctrl := gomock.NewController(t)
			nextgenCli := mock_nextgenglobal.NewMockNextGenGlobal(ctrl)
			auditLogSender := &recordingAuditLogSender{}
			server := NewServer(nextgenCli, auditLogSender)
			nextgenCli.EXPECT().GetResourcePool(gomock.Any(), gomock.Any()).Return(
				testGlobalResourcePoolProfile("pool-a", nil), nil,
			)
			nextgenCli.EXPECT().ResizeResourcePoolCapacity(gomock.Any(), gomock.Any()).DoAndReturn(
				func(_ context.Context, req *globalngv1.ResizeResourcePoolCapacityRequest, _ ...grpc.CallOption) (*globalngv1.ResizeResourcePoolCapacityResponse, error) {
					require.Equal(t, tt.validateOnly, req.GetValidateOnly())
					return &globalngv1.ResizeResourcePoolCapacityResponse{
						CurrentCapacity: testGlobalResourcePoolCapacity(3),
						TargetCapacity:  testGlobalResourcePoolCapacity(5),
					}, nil
				},
			)

			_, err := server.ResizeResourcePoolCapacity(checkedContext(t), &openapingv1beta2.ResizeResourcePoolCapacityRequest{
				ResourcePoolId: "42",
				TargetCapacity: testOpenAPIResourcePoolCapacity(5),
				ValidateOnly:   tt.validateOnly,
			})

			require.NoError(t, err)
			require.Len(t, auditLogSender.records, tt.wantAuditLogs)
			if tt.wantAuditLogs > 0 {
				require.Equal(t, auditlog.EventResizeResourcePoolCapacity, auditLogSender.records[0].eventType)
				require.NoError(t, auditLogSender.records[0].err)
				require.NotNil(t, auditLogSender.records[0].details[auditlog.KeyOldValue])
				require.NotNil(t, auditLogSender.records[0].details[auditlog.KeyNewValue])
			}
		})
	}
}

func TestResolveUpdatePathsDerivesOrValidatesExactlyOneMutableField(t *testing.T) {
	t.Parallel()

	_, _, err := resolveUpdatePaths(&openapingv1beta2.UpdateResourcePoolRequest{
		ResourcePool: &openapingv1beta2.ResourcePool{},
	})
	require.Error(t, err)

	_, _, err = resolveUpdatePaths(&openapingv1beta2.UpdateResourcePoolRequest{
		ResourcePool: &openapingv1beta2.ResourcePool{DisplayName: "pool-a"},
		UpdateMask:   &fieldmaskpb.FieldMask{Paths: []string{"capacity"}},
	})
	require.Error(t, err)

	_, _, err = resolveUpdatePaths(&openapingv1beta2.UpdateResourcePoolRequest{
		ResourcePool: &openapingv1beta2.ResourcePool{
			DisplayName: "pool-a",
			VcpuLimit: &openapingv1beta2.VcpuLimit{
				Policy: &openapingv1beta2.VcpuLimit_Unlimited{Unlimited: true},
			},
		},
	})
	require.Error(t, err)

	updateName, updateVcpu, err := resolveUpdatePaths(&openapingv1beta2.UpdateResourcePoolRequest{
		ResourcePool: &openapingv1beta2.ResourcePool{DisplayName: "pool-a"},
	})
	require.NoError(t, err)
	require.True(t, updateName)
	require.False(t, updateVcpu)
}

func TestProfileToOpenAPIValidateOnlyOmitsAllocatedAndLifecycleFields(t *testing.T) {
	t.Parallel()

	response, err := profileToOpenAPI(&globalngv1.ResourcePoolProfile{
		ResourcePool: &globalngv1.ResourcePool{
			Name:                 "pool-a",
			ProviderId:           globalcommonv1.CloudProvider_AWS,
			RegionId:             globalcommonv1.Region_AWS_US_WEST_2,
			OwnerType:            globalngv1.ResourcePool_ORGANIZATION,
			OwnerId:              1001,
			ServicePlan:          globalcommonv1.ServicePlan_BYOC,
			HighAvailabilityType: globalngv1.ResourcePool_REGIONAL,
			LifecycleState:       globalngv1.ResourcePool_CREATING,
			CreatedBy:            "creator@example.com",
		},
	})
	require.NoError(t, err)
	require.Empty(t, response.GetName())
	require.Empty(t, response.GetResourcePoolId())
	require.Nil(t, response.State)
	require.Nil(t, response.TidbInstanceCount)
	require.Nil(t, response.CreatedBy)
	require.Nil(t, response.CreateTime)
	require.Nil(t, response.UpdateTime)
	require.Nil(t, response.VcpuCapacitySummary)
}

func TestTargetCapacityToGlobalPreservesExplicitZeroTiFlashDisable(t *testing.T) {
	t.Parallel()

	target, err := targetCapacityToGlobal(&openapingv1beta2.ResourcePoolCapacity{
		Tikv: &openapingv1beta2.ComponentCapacity{
			NodeSpec:  &openapingv1beta2.NodeSpecSelection{Cpu: 8, MemoryGi: 32},
			NodeCount: proto.Int64(3),
		},
		Tiflash: &openapingv1beta2.ComponentCapacity{
			NodeCount: proto.Int64(0),
		},
	})
	require.NoError(t, err)
	require.NotNil(t, target.GetTiflash())
	require.NotNil(t, target.GetTiflash().NodeCount)
	require.Zero(t, target.GetTiflash().GetNodeCount())
	require.Nil(t, target.GetTiflash().GetNodeSpec())
}

func TestCapacityConvertersKeepRaftlogInternal(t *testing.T) {
	t.Parallel()

	t.Run("rejects raftlog in customer input", func(t *testing.T) {
		t.Parallel()

		_, err := capacityToGlobal(&openapingv1beta2.ResourcePoolCapacity{
			Tikv: &openapingv1beta2.ComponentCapacity{
				NodeSpec:  &openapingv1beta2.NodeSpecSelection{Cpu: 8, MemoryGi: 32},
				NodeCount: proto.Int64(3),
				Storages: []*openapingv1beta2.Storage{
					{Name: "data", SizeGi: 500},
					{Name: "raftlog", SizeGi: 100},
				},
			},
		})

		require.ErrorContains(t, err, `unsupported storage "raftlog"`)
	})

	t.Run("omits raftlog from capacity output", func(t *testing.T) {
		t.Parallel()

		capacity := capacityToOpenAPI(&globalngv1.ResourcePoolCapacity{
			Tikv: &globalngv1.ComponentCapacity{
				Storages: []*globalngv1.Storage{
					{Name: "data", SizeGi: 500},
					{Name: "raftlog", SizeGi: 100},
				},
			},
		})

		require.Equal(t, []*openapingv1beta2.Storage{{Name: "data", SizeGi: 500}},
			capacity.GetTikv().GetStorages())
	})

	t.Run("omits raftlog from node spec output", func(t *testing.T) {
		t.Parallel()

		nodeSpec := nodeSpecToOpenAPI(&globalngv1.ResourcePoolNodeSpec{
			Storages: []*globalngv1.NodeSpecStorage{
				{Name: "data", DefaultSizeGi: 500},
				{Name: "raftlog", DefaultSizeGi: 100},
			},
		})

		require.Equal(t, []*openapingv1beta2.NodeSpecStorage{{Name: "data", DefaultSizeGi: 500}},
			nodeSpec.GetStorages())
	})
}

func TestServerUpsertResourcePoolCloudResourceTagsForwardsRemovals(t *testing.T) {
	t.Parallel()

	ctrl := gomock.NewController(t)
	nextgenCli := mock_nextgenglobal.NewMockNextGenGlobal(ctrl)
	auditLogSender := &recordingAuditLogSender{}
	server := NewServer(nextgenCli, auditLogSender)
	ctx := checkedContext(t)

	nextgenCli.EXPECT().GetResourcePool(gomock.Any(), gomock.Any()).Return(
		testGlobalResourcePoolProfile("pool-a", map[string]string{"deprecated": "old-secret"}), nil,
	)
	nextgenCli.EXPECT().UpsertResourcePoolCloudResourceTags(gomock.Any(), gomock.Any()).DoAndReturn(
		func(_ context.Context, request *globalngv1.UpsertResourcePoolCloudResourceTagsRequest, _ ...grpc.CallOption) (*globalngv1.UpsertResourcePoolCloudResourceTagsResponse, error) {
			require.Equal(t, uint64(42), request.GetResourcePoolId())
			require.Equal(t, uint64(1001), request.GetOrganizationId())
			require.Equal(t, map[string]string{"owner": "platform"}, request.GetCloudResourceTags())
			require.Equal(t, []string{"deprecated"}, request.GetRemoveCloudResourceTagKeys())
			return &globalngv1.UpsertResourcePoolCloudResourceTagsResponse{
				CloudResourceTags: map[string]string{"owner": "platform"},
			}, nil
		},
	)

	response, err := server.UpsertResourcePoolCloudResourceTags(
		ctx,
		&openapingv1beta2.UpsertResourcePoolCloudResourceTagsRequest{
			ResourcePoolId:             "42",
			CloudResourceTags:          map[string]string{"owner": "platform"},
			RemoveCloudResourceTagKeys: []string{"deprecated"},
		},
	)

	require.NoError(t, err)
	require.Equal(t, map[string]string{"owner": "platform"}, response.GetCloudResourceTags())
	require.Len(t, auditLogSender.records, 1)
	record := auditLogSender.records[0]
	require.Equal(t, auditlog.EventUpdateResourcePoolCloudResourceTags, record.eventType)
	require.Equal(t, []string{"owner"}, record.details[auditlog.KeyChangedTagKeys])
	require.Equal(t, []string{"deprecated"}, record.details[auditlog.KeyRemovedTagKeys])
	require.Equal(t, 1, record.details[auditlog.KeyOldTagCount])
	require.Equal(t, 1, record.details[auditlog.KeyNewTagCount])
	detailJSON, err := json.Marshal(record.details)
	require.NoError(t, err)
	require.NotContains(t, string(detailJSON), "platform")
	require.NotContains(t, string(detailJSON), "old-secret")
}

func testGlobalResourcePoolProfile(name string, tags map[string]string) *globalngv1.ResourcePoolProfile {
	return &globalngv1.ResourcePoolProfile{
		ResourcePool: &globalngv1.ResourcePool{
			Id:                   42,
			Name:                 name,
			ProviderId:           globalcommonv1.CloudProvider_AWS,
			RegionId:             globalcommonv1.Region_AWS_US_WEST_2,
			OwnerType:            globalngv1.ResourcePool_ORGANIZATION,
			OwnerId:              1001,
			Cidr:                 "10.0.0.0/16",
			ServicePlan:          globalcommonv1.ServicePlan_BYOC,
			LifecycleState:       globalngv1.ResourcePool_ACTIVE,
			HighAvailabilityType: globalngv1.ResourcePool_ZONAL,
			VcpusLimit:           120,
			VcpuLimitMode:        globalngv1.ResourcePool_LIMITED,
			CloudResourceTags:    tags,
		},
		Region:   &globalcommonv1.Region{DisplayName: "Oregon (us-west-2)"},
		Capacity: testGlobalResourcePoolCapacity(3),
	}
}

func testGlobalResourcePoolCapacity(nodeCount int64) *globalngv1.ResourcePoolCapacity {
	return &globalngv1.ResourcePoolCapacity{
		Tikv: &globalngv1.ComponentCapacity{
			NodeSpec:  &globalngv1.NodeSpecSelection{Cpu: 8, MemoryGi: 32},
			NodeCount: proto.Int64(nodeCount),
			Storages:  []*globalngv1.Storage{{Name: "data", SizeGi: 500}},
		},
	}
}

func testOpenAPIResourcePoolCapacity(nodeCount int64) *openapingv1beta2.ResourcePoolCapacity {
	return &openapingv1beta2.ResourcePoolCapacity{
		Tikv: &openapingv1beta2.ComponentCapacity{
			NodeSpec:  &openapingv1beta2.NodeSpecSelection{Cpu: 8, MemoryGi: 32},
			NodeCount: proto.Int64(nodeCount),
			Storages:  []*openapingv1beta2.Storage{{Name: "data", SizeGi: 500}},
		},
	}
}

func checkedContext(t *testing.T) context.Context {
	t.Helper()
	return metadata.AppendToIncomingMeta(t.Context(), &metadata.RequestMeta{
		OrgID:             1001,
		UserID:            "user-1",
		Email:             "creator@example.com",
		DefaultProjectID:  2001,
		ValidProjectList:  []uint64{2001},
		IsCheckPermission: proto.Bool(true),
	})
}

type recordedAuditLog struct {
	eventType auditlog.EventType
	reqMeta   *metadata.RequestMeta
	details   auditlog.Details
	err       error
}

type recordingAuditLogSender struct {
	records []recordedAuditLog
}

func (s *recordingAuditLogSender) SubmitAuditLog(
	context.Context,
	auditlog.EventType,
	*metadata.RequestMeta,
	func() auditlog.Details,
	error,
) {
	panic("resource pool audit logs must be organization-scoped")
}

func (s *recordingAuditLogSender) SubmitAuditLogForNextgen(
	_ context.Context,
	eventType auditlog.EventType,
	reqMeta *metadata.RequestMeta,
	detailsFunc func() auditlog.Details,
	err error,
) {
	s.records = append(s.records, recordedAuditLog{
		eventType: eventType,
		reqMeta:   reqMeta,
		details:   detailsFunc(),
		err:       err,
	})
}

var _ eventbus.ConsoleAuditLogSender = (*recordingAuditLogSender)(nil)