package sharedpool

import (
	"context"
	"testing"
	"time"

	v1alpha1br "github.com/pingcap/tidb-operator/api/v2/br/v1alpha1"
	"github.com/pingcap/tidb-operator/api/v2/core/v1alpha1"
	"github.com/stretchr/testify/mock"
	"github.com/stretchr/testify/require"
	apiv1 "github.com/tidbcloud/cluster-service-ng/api/cluster-service-ng/v1beta"
	"github.com/tidbcloud/cluster-service-ng/domain/entity"
	dataplanemocks "github.com/tidbcloud/cluster-service-ng/pkg/dataplane/mocks"
	schemav1 "github.com/tidbcloud/cluster-service-ng/pkg/storage/etcd/schema/v1"
	"github.com/tidbcloud/cluster-service-ng/pkg/storage/mysql/schema"
	repomocks "github.com/tidbcloud/cluster-service-ng/pkg/storage/repo/mocks"
	"google.golang.org/grpc/codes"
	"google.golang.org/grpc/status"
	"google.golang.org/protobuf/encoding/protojson"
	"gorm.io/gorm"
	corev1 "k8s.io/api/core/v1"
	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
	"k8s.io/apimachinery/pkg/runtime"
	clientgoscheme "k8s.io/client-go/kubernetes/scheme"
	"sigs.k8s.io/controller-runtime/pkg/client/fake"
)

func TestCvtSharedPoolToModelCopiesPlacementGroupEnabled(t *testing.T) {
	r := &sharedPoolRepoImpl{}

	model, _, err := r.cvtSharedPoolToModel(&entity.SharedPool{SharedPool: apiv1.SharedPool{
		Id:            "sp-1",
		DataplaneName: "dp-1",
		Provider: &apiv1.Provider{
			PlacementGroupEnabled: true,
		},
		ObjectStorage: &entity.ObjectStorage{Bucket: "bucket", Region: "us-west-2"},
	}})

	require.NoError(t, err)
	require.True(t, model.PlacementGroupEnabled)
}

func TestCvtSharedPoolCapacityPlannerRoundTrip(t *testing.T) {
	r := &sharedPoolRepoImpl{}
	model, _, err := r.cvtSharedPoolToModel(&entity.SharedPool{
		SharedPool: apiv1.SharedPool{
			Id:            "sp-1",
			DataplaneName: "dp-1",
			Provider:      &apiv1.Provider{},
			ObjectStorage: &entity.ObjectStorage{Bucket: "bucket", Region: "us-west-2"},
		},
		EnableCapacityPlanner: true,
	})
	require.NoError(t, err)
	require.True(t, model.EnableCapacityPlanner)

	model.DeletedAt = gorm.DeletedAt{Time: time.Now(), Valid: true}
	sharedPool, err := r.cvtSharedPoolToEn(context.Background(), model)
	require.NoError(t, err)
	require.True(t, sharedPool.EnableCapacityPlanner)
}

func TestCvtSharedPoolToModelStoresCloudResourceTags(t *testing.T) {
	r := &sharedPoolRepoImpl{}

	model, _, err := r.cvtSharedPoolToModel(&entity.SharedPool{SharedPool: apiv1.SharedPool{
		Id:            "sp-1",
		DataplaneName: "dp-1",
		Provider:      &apiv1.Provider{},
		ObjectStorage: &entity.ObjectStorage{Bucket: "bucket", Region: "us-west-2"},
		CloudResourceTags: map[string]string{
			"cost-center": "cc-1",
			"owner":       "platform",
		},
	}})

	require.NoError(t, err)
	require.JSONEq(t, `{
		"cost-center": "cc-1",
		"owner": "platform"
	}`, model.CloudResourceTags)
}

func TestUnmarshalStringMapDefaultsBlankAndNullToEmptyMap(t *testing.T) {
	tests := []struct {
		name string
		raw  string
	}{
		{
			name: "blank",
			raw:  "",
		},
		{
			name: "null",
			raw:  "null",
		},
	}

	for _, tt := range tests {
		t.Run(tt.name, func(t *testing.T) {
			tags, err := unmarshalStringMap(tt.raw)

			require.NoError(t, err)
			require.NotNil(t, tags)
			require.Empty(t, tags)
		})
	}
}

func TestCvtSharedPoolPreservesConfigCenterMetaProfileID(t *testing.T) {
	r := &sharedPoolRepoImpl{}

	model, _, err := r.cvtSharedPoolToModel(&entity.SharedPool{
		SharedPool: apiv1.SharedPool{
			Id:            "sp-1",
			DataplaneName: "dp-1",
			Provider:      &apiv1.Provider{},
			ObjectStorage: &entity.ObjectStorage{Bucket: "bucket", Region: "us-west-2"},
		},
		ConfigCenterMetaProfileId: "1001",
	})

	require.NoError(t, err)
	require.Equal(t, "1001", model.ConfigCenterMetaProfileID)

	ctx := context.Background()
	dataplaneRepo := repomocks.NewDataplaneRepo(t)
	dataplaneRepo.On("LoadByName", mock.Anything, "dp-1").Return(nil, status.Error(codes.NotFound, "not found")).Once()
	r.dataplaneRepo = dataplaneRepo

	sharedPool, err := r.cvtSharedPoolToEn(ctx, model)

	require.NoError(t, err)
	require.Equal(t, "1001", sharedPool.ConfigCenterMetaProfileId)
}

func TestCvtSharedPoolPreservesEmptyConfigCenterMetaProfileID(t *testing.T) {

	tests := []struct {
		name      string
		profileID string
	}{
		{name: "no profile", profileID: ""},
	}

	for _, tt := range tests {
		t.Run(tt.name, func(t *testing.T) {
			r := &sharedPoolRepoImpl{}
			model, _, err := r.cvtSharedPoolToModel(&entity.SharedPool{
				SharedPool: apiv1.SharedPool{
					Id:            "sp-1",
					DataplaneName: "dp-1",
					Provider:      &apiv1.Provider{},
					ObjectStorage: &entity.ObjectStorage{Bucket: "bucket", Region: "us-west-2"},
				},
				ConfigCenterMetaProfileId: tt.profileID,
			})

			require.NoError(t, err)
			require.Equal(t, tt.profileID, model.ConfigCenterMetaProfileID)

			dataplaneRepo := repomocks.NewDataplaneRepo(t)
			dataplaneRepo.On("LoadByName", mock.Anything, "dp-1").
				Return(nil, status.Error(codes.NotFound, "not found")).Once()
			r.dataplaneRepo = dataplaneRepo
			sharedPool, err := r.cvtSharedPoolToEn(context.Background(), model)

			require.NoError(t, err)
			require.Equal(t, tt.profileID, sharedPool.ConfigCenterMetaProfileId)
		})
	}
}

func TestCvtSharedPoolToModelStoresHighAvailabilityType(t *testing.T) {
	r := &sharedPoolRepoImpl{}

	tests := []struct {
		name   string
		haType apiv1.SharedPool_HighAvailabilityType
		want   string
	}{
		{
			name:   "unspecified is preserved",
			haType: apiv1.SharedPool_HIGH_AVAILABILITY_TYPE_UNSPECIFIED,
			want:   apiv1.SharedPool_HIGH_AVAILABILITY_TYPE_UNSPECIFIED.String(),
		},
		{
			name:   "zonal is preserved",
			haType: apiv1.SharedPool_ZONAL,
			want:   apiv1.SharedPool_ZONAL.String(),
		},
		{
			name:   "regional is preserved",
			haType: apiv1.SharedPool_REGIONAL,
			want:   apiv1.SharedPool_REGIONAL.String(),
		},
	}

	for _, tt := range tests {
		t.Run(tt.name, func(t *testing.T) {
			model, _, err := r.cvtSharedPoolToModel(&entity.SharedPool{
				SharedPool: apiv1.SharedPool{
					Id:                   "sp-1",
					DataplaneName:        "dp-1",
					Provider:             &apiv1.Provider{},
					ObjectStorage:        &entity.ObjectStorage{Bucket: "bucket", Region: "us-west-2"},
					HighAvailabilityType: tt.haType,
				},
			})

			require.NoError(t, err)
			require.Equal(t, tt.want, model.HighAvailabilityType)
		})
	}
}

func TestCvtSharedPoolTiflashModeRoundTrip(t *testing.T) {
	t.Run("entity to model", func(t *testing.T) {
		r := &sharedPoolRepoImpl{}
		model, _, err := r.cvtSharedPoolToModel(&entity.SharedPool{
			SharedPool: apiv1.SharedPool{
				Id:            "sp-1",
				DataplaneName: "dp-1",
				Provider:      &apiv1.Provider{},
				ObjectStorage: &entity.ObjectStorage{Bucket: "bucket", Region: "us-west-2"},
			},
			TiflashMode: schema.SharedPoolTiflashModeColumnar,
		})

		require.NoError(t, err)
		require.Equal(t, schema.SharedPoolTiflashModeColumnar, model.TiflashMode)
	})

	t.Run("deleted model to entity", func(t *testing.T) {
		r := &sharedPoolRepoImpl{}
		sp, err := r.cvtSharedPoolToEn(context.Background(), &schema.SharedPool{
			ID:          "sp-1",
			TiflashMode: schema.SharedPoolTiflashModeColumnar,
			DeletedAt:   gorm.DeletedAt{Time: time.Now(), Valid: true},
		})

		require.NoError(t, err)
		require.Equal(t, schema.SharedPoolTiflashModeColumnar, sp.TiflashMode)
	})

	t.Run("provider not ready model to entity", func(t *testing.T) {
		dataplaneRepo := repomocks.NewDataplaneRepo(t)
		dataplaneRepo.On("LoadByName", mock.Anything, "dp-1").Return(nil, status.Error(codes.NotFound, "not found")).Once()
		r := &sharedPoolRepoImpl{dataplaneRepo: dataplaneRepo}

		sp, err := r.cvtSharedPoolToEn(context.Background(), &schema.SharedPool{
			ID:            "sp-1",
			DataplaneName: "dp-1",
			TiflashMode:   schema.SharedPoolTiflashModeColumnar,
		})

		require.NoError(t, err)
		require.Equal(t, schema.SharedPoolTiflashModeColumnar, sp.TiflashMode)
	})

	t.Run("normal load model to entity", func(t *testing.T) {
		const (
			sharedPoolID  = "sp-1"
			dataplaneName = "dp-1"
		)

		scheme := runtime.NewScheme()
		require.NoError(t, clientgoscheme.AddToScheme(scheme))
		require.NoError(t, v1alpha1.AddToScheme(scheme))
		require.NoError(t, v1alpha1br.AddToScheme(scheme))

		cluster := &v1alpha1.Cluster{
			ObjectMeta: metav1.ObjectMeta{
				Name:      "db",
				Namespace: "tidb-sp-1",
			},
			Status: v1alpha1.ClusterStatus{
				ObservedGeneration: 1,
				Conditions: []metav1.Condition{{
					Type:   v1alpha1.ClusterCondAvailable,
					Status: metav1.ConditionFalse,
				}},
			},
		}
		cluster.Generation = 1
		remoteClient := fake.NewClientBuilder().WithScheme(scheme).WithObjects(cluster).Build()

		dataplaneRepo := repomocks.NewDataplaneRepo(t)
		dataplaneRepo.On("LoadByName", mock.Anything, dataplaneName).Return(&entity.DataPlaneProvider{
			Name: dataplaneName,
		}, nil).Once()
		dataplaneManager := dataplanemocks.NewManager(t)
		dataplaneManager.On("GetClient", dataplaneName).Return(remoteClient, nil).Once()
		backupGcSettingRepo := repomocks.NewBackupGcSettingRepo(t)
		backupGcSettingRepo.On("Load", mock.Anything, sharedPoolID).Return(nil, status.Error(codes.NotFound, "not found")).Once()

		r := &sharedPoolRepoImpl{
			dataplaneRepo:       dataplaneRepo,
			dataplaneManager:    dataplaneManager,
			backupGcSettingRepo: backupGcSettingRepo,
		}
		sp, err := r.cvtSharedPoolToEn(context.Background(), &schema.SharedPool{
			ID:            sharedPoolID,
			DataplaneName: dataplaneName,
			TiflashMode:   schema.SharedPoolTiflashModeColumnar,
		})

		require.NoError(t, err)
		require.Equal(t, schema.SharedPoolTiflashModeColumnar, sp.TiflashMode)
	})
}

func TestCvtSharedPoolToModelNormalizesEnabledTiKVRemoteWorkers(t *testing.T) {
	r := &sharedPoolRepoImpl{}

	tests := []struct {
		name           string
		capacityMode   apiv1.CapacityMode
		enabledWorkers []apiv1.ComponentType
		want           string
	}{
		{
			name: "empty defaults to tikv and coprocessor",
			want: "COPROCESSOR_WORKER,TIKV_WORKER",
		},
		{
			name: "worker only is preserved",
			enabledWorkers: []apiv1.ComponentType{
				apiv1.ComponentType_TIKV_WORKER,
			},
			want: "TIKV_WORKER",
		},
		{
			name: "duplicates are deduplicated and sorted",
			enabledWorkers: []apiv1.ComponentType{
				apiv1.ComponentType_TIKV_WORKER,
				apiv1.ComponentType_TIKV_WORKER,
				apiv1.ComponentType_COPROCESSOR_WORKER,
			},
			want: "COPROCESSOR_WORKER,TIKV_WORKER",
		},
		{
			name:         "topology driven stores empty string",
			capacityMode: apiv1.CapacityMode_TOPOLOGY_DRIVEN,
			want:         "",
		},
	}

	for _, tt := range tests {
		t.Run(tt.name, func(t *testing.T) {
			model, _, err := r.cvtSharedPoolToModel(&entity.SharedPool{SharedPool: apiv1.SharedPool{
				Id:                       "sp-1",
				DataplaneName:            "dp-1",
				CapacityMode:             tt.capacityMode,
				EnabledTikvRemoteWorkers: tt.enabledWorkers,
				Provider:                 &apiv1.Provider{},
				ObjectStorage:            &entity.ObjectStorage{Bucket: "bucket", Region: "us-west-2"},
			}})

			require.NoError(t, err)
			require.Equal(t, tt.want, model.EnabledTiKVRemoteWorkers)
		})
	}
}

func TestCvtSharedPoolToModelRejectsInvalidCapacityMode(t *testing.T) {
	r := &sharedPoolRepoImpl{}

	model, stores, err := r.cvtSharedPoolToModel(&entity.SharedPool{SharedPool: apiv1.SharedPool{
		Id:                       "sp-1",
		DataplaneName:            "dp-1",
		CapacityMode:             apiv1.CapacityMode(99),
		EnabledTikvRemoteWorkers: []apiv1.ComponentType{apiv1.ComponentType_TIKV_WORKER},
		Provider:                 &apiv1.Provider{},
		ObjectStorage:            &entity.ObjectStorage{Bucket: "bucket", Region: "us-west-2"},
	}})

	require.Error(t, err)
	require.Nil(t, model)
	require.Nil(t, stores)
	require.Contains(t, err.Error(), "invalid shared pool capacity mode")
}

func TestCvtSharedPoolToEnRestoresPlacementGroupEnabledWhenProviderNotReady(t *testing.T) {
	ctx := context.Background()
	dataplaneRepo := repomocks.NewDataplaneRepo(t)
	dataplaneRepo.On("LoadByName", mock.Anything, "dp-1").Return(&entity.DataPlaneProvider{
		Tags: map[string]string{schemav1.LabelUserTenantKey: "12345"},
	}, status.Error(codes.NotFound, "not ready")).Once()
	r := &sharedPoolRepoImpl{dataplaneRepo: dataplaneRepo}

	sp, err := r.cvtSharedPoolToEn(ctx, &schema.SharedPool{
		ID:                    "sp-1",
		DataplaneName:         "dp-1",
		PlacementGroupEnabled: true,
	})

	require.NoError(t, err)
	require.True(t, sp.GetProvider().GetPlacementGroupEnabled())
	require.Equal(t, "12345", sp.GetProvider().GetTags()[schemav1.LabelUserTenantKey])
	require.Empty(t, sp.TenantId)
}

func TestCvtSharedPoolToEnPreservesTenantTagFromActiveProvider(t *testing.T) {
	ctx := context.Background()
	provider := &entity.DataPlaneProvider{
		Name: "dp-1",
		Tags: map[string]string{schemav1.LabelUserTenantKey: "12345"},
	}
	dataplaneRepo := repomocks.NewDataplaneRepo(t)
	dataplaneRepo.On("LoadByName", mock.Anything, "dp-1").Return(provider, nil).Once()

	scheme := runtime.NewScheme()
	require.NoError(t, corev1.AddToScheme(scheme))
	require.NoError(t, v1alpha1.AddToScheme(scheme))
	dataplaneManager := dataplanemocks.NewManager(t)
	dataplaneManager.On("GetClient", "dp-1").Return(fake.NewClientBuilder().WithScheme(scheme).Build(), nil).Once()
	backupGcSettingRepo := repomocks.NewBackupGcSettingRepo(t)
	backupGcSettingRepo.On("Load", mock.Anything, "sp-1").
		Return(nil, status.Error(codes.NotFound, "not found")).Once()

	r := &sharedPoolRepoImpl{
		dataplaneRepo:       dataplaneRepo,
		dataplaneManager:    dataplaneManager,
		backupGcSettingRepo: backupGcSettingRepo,
	}
	sp, err := r.cvtSharedPoolToEn(ctx, &schema.SharedPool{
		ID:            "sp-1",
		DataplaneName: "dp-1",
	})

	require.NoError(t, err)
	require.Same(t, provider, sp.GetProvider())
	require.Equal(t, "12345", sp.GetProvider().GetTags()[schemav1.LabelUserTenantKey])
	require.Empty(t, sp.TenantId)
}

func TestCvtSharedPoolToEnRestoresCloudResourceTagsWhenProviderNotReady(t *testing.T) {
	ctx := context.Background()
	dataplaneRepo := repomocks.NewDataplaneRepo(t)
	dataplaneRepo.On("LoadByName", mock.Anything, "dp-1").Return(nil, status.Error(codes.NotFound, "not found")).Once()
	r := &sharedPoolRepoImpl{dataplaneRepo: dataplaneRepo}

	sp, err := r.cvtSharedPoolToEn(ctx, &schema.SharedPool{
		ID:                "sp-1",
		DataplaneName:     "dp-1",
		CloudResourceTags: `{"owner":"new-owner","tags.tidbcloud.com/cost-center":"cc-1","tags.tidbcloud.com/owner":"legacy-owner"}`,
	})

	require.NoError(t, err)
	require.Equal(t, map[string]string{
		"cost-center": "cc-1",
		"owner":       "new-owner",
	}, sp.GetCloudResourceTags())
}

func TestCvtSharedPoolToEnRestoresHighAvailabilityTypeWhenProviderNotReady(t *testing.T) {
	tests := []struct {
		name  string
		value string
		want  apiv1.SharedPool_HighAvailabilityType
	}{
		{
			name: "empty is unspecified",
			want: apiv1.SharedPool_HIGH_AVAILABILITY_TYPE_UNSPECIFIED,
		},
		{
			name:  "zonal is restored",
			value: apiv1.SharedPool_ZONAL.String(),
			want:  apiv1.SharedPool_ZONAL,
		},
		{
			name:  "regional is restored",
			value: apiv1.SharedPool_REGIONAL.String(),
			want:  apiv1.SharedPool_REGIONAL,
		},
		{
			name:  "unknown stored value is unspecified",
			value: "INVALID",
			want:  apiv1.SharedPool_HIGH_AVAILABILITY_TYPE_UNSPECIFIED,
		},
	}

	for _, tt := range tests {
		t.Run(tt.name, func(t *testing.T) {
			ctx := context.Background()
			dataplaneRepo := repomocks.NewDataplaneRepo(t)
			dataplaneRepo.On("LoadByName", mock.Anything, "dp-1").Return(nil, status.Error(codes.NotFound, "not found")).Once()
			r := &sharedPoolRepoImpl{dataplaneRepo: dataplaneRepo}

			sp, err := r.cvtSharedPoolToEn(ctx, &schema.SharedPool{
				ID:                   "sp-1",
				DataplaneName:        "dp-1",
				HighAvailabilityType: tt.value,
			})

			require.NoError(t, err)
			require.Equal(t, tt.want, sp.GetHighAvailabilityType())
		})
	}
}

func TestCvtSharedPoolToEnRejectsInvalidCapacityMode(t *testing.T) {
	r := &sharedPoolRepoImpl{}

	sp, err := r.cvtSharedPoolToEn(context.Background(), &schema.SharedPool{
		ID:           "sp-1",
		CapacityMode: "INVALID_CAPACITY_MODE",
	})

	require.Error(t, err)
	require.Nil(t, sp)
	require.Contains(t, err.Error(), "invalid shared pool capacity mode")
}

func TestCvtSharedPoolToEnRestoresEnabledTiKVRemoteWorkersWhenProviderNotReady(t *testing.T) {
	tests := []struct {
		name                     string
		capacityMode             string
		enabledTiKVRemoteWorkers string
		want                     []apiv1.ComponentType
	}{
		{
			name:                     "unspecified capacity mode is treated as rcu driven",
			enabledTiKVRemoteWorkers: "TIKV_WORKER",
			want:                     []apiv1.ComponentType{apiv1.ComponentType_TIKV_WORKER},
		},
		{
			name:                     "capacity mode ignores surrounding whitespace",
			capacityMode:             " " + apiv1.CapacityMode_RCU_DRIVEN.String() + " ",
			enabledTiKVRemoteWorkers: "TIKV_WORKER",
			want:                     []apiv1.ComponentType{apiv1.ComponentType_TIKV_WORKER},
		},
		{
			name:                     "topology driven returns empty list",
			capacityMode:             apiv1.CapacityMode_TOPOLOGY_DRIVEN.String(),
			enabledTiKVRemoteWorkers: "COPROCESSOR_WORKER,TIKV_WORKER",
			want:                     []apiv1.ComponentType{},
		},
		{
			name:                     "invalid stored workers fall back to default workers",
			capacityMode:             apiv1.CapacityMode_RCU_DRIVEN.String(),
			enabledTiKVRemoteWorkers: "COPROCESSOR_WORKER",
			want: []apiv1.ComponentType{
				apiv1.ComponentType_COPROCESSOR_WORKER,
				apiv1.ComponentType_TIKV_WORKER,
			},
		},
	}

	for _, tt := range tests {
		t.Run(tt.name, func(t *testing.T) {
			ctx := context.Background()
			dataplaneRepo := repomocks.NewDataplaneRepo(t)
			dataplaneRepo.On("LoadByName", mock.Anything, "dp-1").Return(nil, status.Error(codes.NotFound, "not found")).Once()
			r := &sharedPoolRepoImpl{dataplaneRepo: dataplaneRepo}

			sp, err := r.cvtSharedPoolToEn(ctx, &schema.SharedPool{
				ID:                       "sp-1",
				DataplaneName:            "dp-1",
				CapacityMode:             tt.capacityMode,
				EnabledTiKVRemoteWorkers: tt.enabledTiKVRemoteWorkers,
			})

			require.NoError(t, err)
			require.Equal(t, tt.want, sp.GetEnabledTikvRemoteWorkers())
		})
	}
}

// TestCvtSharedPoolHighAvailabilityTypeRoundTrip pins the canonical-state invariant for
// issue #3035: a SharedPool's HA type must be the single embedded apiv1.SharedPool.HighAvailabilityType,
// and one DB round trip (entity → model → entity) must preserve it without divergence.
func TestCvtSharedPoolHighAvailabilityTypeRoundTrip(t *testing.T) {
	tests := []struct {
		name string
		ha   apiv1.SharedPool_HighAvailabilityType
	}{
		{name: "zonal", ha: apiv1.SharedPool_ZONAL},
		{name: "regional", ha: apiv1.SharedPool_REGIONAL},
	}

	for _, tt := range tests {
		t.Run(tt.name, func(t *testing.T) {
			ctx := context.Background()
			dataplaneRepo := repomocks.NewDataplaneRepo(t)
			dataplaneRepo.On("LoadByName", mock.Anything, "dp-1").Return(nil, status.Error(codes.NotFound, "not found")).Once()
			r := &sharedPoolRepoImpl{dataplaneRepo: dataplaneRepo}

			original := &entity.SharedPool{
				SharedPool: apiv1.SharedPool{
					Id:                   "sp-1",
					DataplaneName:        "dp-1",
					Provider:             &apiv1.Provider{},
					ObjectStorage:        &entity.ObjectStorage{Bucket: "bucket", Region: "us-west-2"},
					HighAvailabilityType: tt.ha,
				},
			}

			model, _, err := r.cvtSharedPoolToModel(original)
			require.NoError(t, err)
			require.Equal(t, tt.ha.String(), model.HighAvailabilityType,
				"persisted string must be the canonical enum name")

			loaded, err := r.cvtSharedPoolToEn(ctx, model)
			require.NoError(t, err)
			require.Equal(t, tt.ha, loaded.GetHighAvailabilityType(),
				"hydrated canonical value must equal the original")
		})
	}
}

// TestCvtSharedPoolProtojsonRoundTripForCreateBucketAction is the focused regression test
// for the create-bucket DAG action that PR #3042 was opened to fix. CreateSharedPool freezes
// the entity into workflow.SharedPoolKey via protojson.Marshal, and CreateBucketAction.Start
// reverses that via protojson.Unmarshal into another *entity.SharedPool. Before the canonical
// state was introduced, the entity carried two Go fields with the same name (the outer
// request-enum field plus the embedded apiv1.SharedPool.HighAvailabilityType), and the
// pre-canonicalization PR #2964 + a follow-up rename forced protojson through a path that
// produced an "unknown field highAvailabilityType" rejection during this round trip for
// Premium REGIONAL clusters. The test pins the post-canonicalization invariant: only the
// embedded proto field is present, and the JSON the executor DAG sees is the single,
// canonical highAvailabilityType field.
func TestCvtSharedPoolProtojsonRoundTripForCreateBucketAction(t *testing.T) {
	tests := []struct {
		name string
		ha   apiv1.SharedPool_HighAvailabilityType
	}{
		{name: "zonal", ha: apiv1.SharedPool_ZONAL},
		{name: "regional", ha: apiv1.SharedPool_REGIONAL},
	}

	for _, tt := range tests {
		t.Run(tt.name, func(t *testing.T) {
			original := &entity.SharedPool{
				SharedPool: apiv1.SharedPool{
					Id:                   "sp-1",
					DataplaneName:        "dp-1",
					Provider:             &apiv1.Provider{},
					ObjectStorage:        &entity.ObjectStorage{Bucket: "bucket", Region: "us-west-2"},
					HighAvailabilityType: tt.ha,
				},
				TenantId: "test-tenant",
			}

			data, err := protojson.Marshal(original)
			require.NoError(t, err, "Create path: protojson.Marshal must succeed")

			var loaded entity.SharedPool
			err = protojson.Unmarshal(data, &loaded)
			require.NoError(t, err,
				"CreateBucketAction path: protojson.Unmarshal into *entity.SharedPool must succeed — "+
					"regression guard for the create-bucket failure this PR fixes")
			require.Equal(t, tt.ha, loaded.GetHighAvailabilityType(),
				"canonical HA value must survive the DAG-arg round trip")
			require.NotContains(t, string(data), "inputHighAvailabilityType",
				"the outer request-enum shadow field must not appear in the executor DAG JSON")
			require.Contains(t, string(data), "highAvailabilityType",
				"the canonical embedded field must appear in the executor DAG JSON")
		})
	}
}