package regionalserver

import (
	"testing"

	clusterngv1beta "github.com/tidbcloud/cluster-service-ng/api/cluster-service-ng/v1beta"
	"google.golang.org/genproto/googleapis/api/annotations"
	"google.golang.org/protobuf/proto"
	"google.golang.org/protobuf/reflect/protoreflect"
)

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

	fields := (&clusterngv1beta.CreateSharedPoolRequest{}).ProtoReflect().Descriptor().Fields()
	for name, number := range map[protoreflect.Name]protoreflect.FieldNumber{
		"tiflash_mode":                     23,
		"enforce_o11y_network_non_overlap": 24,
		"bootstrap_capacity":               25,
		"enable_capacity_planner":          26,
	} {
		field := fields.ByName(name)
		if field == nil {
			t.Fatalf("CreateSharedPoolRequest.%s not found", name)
		}
		if field.Number() != number {
			t.Errorf("CreateSharedPoolRequest.%s field number = %d, want %d", name, field.Number(), number)
		}
		if field.ContainingOneof() != nil {
			t.Errorf("CreateSharedPoolRequest.%s must remain outside capacity_spec", name)
		}
	}

	assertCSNGMethod(t, "NodeInstanceGroupService", "ListNodeSpecs")
	assertCSNGMethod(t, "SharedPoolService", "GetSharedPoolProvisionConfiguration")
	evaluate := assertCSNGMethod(t, "SharedPoolService", "EvaluateSharedPoolNetworkCapacity")
	httpRule, ok := proto.GetExtension(evaluate.Options(), annotations.E_Http).(*annotations.HttpRule)
	if !ok || httpRule.GetPost() != "/sharedPools:evaluateNetworkCapacity" || httpRule.GetBody() != "*" {
		t.Fatalf("EvaluateSharedPoolNetworkCapacity HTTP route = %v", httpRule)
	}
	requestFields := evaluate.Input().Fields()
	if requestFields.Len() != 3 {
		t.Fatalf("EvaluateSharedPoolNetworkCapacity request has %d fields, want 3", requestFields.Len())
	}
	prefix := requestFields.ByName("cidr_prefix_length")
	if prefix == nil || prefix.Number() != 1 || prefix.Kind() != protoreflect.Int32Kind || !prefix.HasPresence() {
		t.Errorf("cidr_prefix_length descriptor = %v, want optional int32 field 1", prefix)
	}
	ha := requestFields.ByName("high_availability_type")
	if ha == nil || ha.Number() != 2 || ha.Kind() != protoreflect.EnumKind {
		t.Errorf("high_availability_type descriptor = %v, want enum field 2", ha)
	}
	nodeZones := requestFields.ByName("node_zones")
	if nodeZones == nil || nodeZones.Number() != 3 || nodeZones.Kind() != protoreflect.StringKind ||
		nodeZones.Cardinality() != protoreflect.Repeated || nodeZones.HasPresence() {
		t.Errorf("node_zones descriptor = %v, want repeated string field 3 without presence", nodeZones)
	}
	responseFields := evaluate.Output().Fields()
	supported := responseFields.ByName("supported")
	supportedNodeCount := responseFields.ByName("supported_node_count")
	if responseFields.Len() != 2 || supported == nil || supported.Kind() != protoreflect.BoolKind ||
		supportedNodeCount == nil || supportedNodeCount.Kind() != protoreflect.Int64Kind {
		t.Errorf("EvaluateSharedPoolNetworkCapacity response fields are incompatible")
	}
}

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

	tests := []struct {
		name     string
		wire     []byte
		validate func(*testing.T, *clusterngv1beta.CreateSharedPoolRequest)
	}{
		{
			name: "request unit remains field 2",
			wire: []byte{0x12, 0x00},
			validate: func(t *testing.T, request *clusterngv1beta.CreateSharedPoolRequest) {
				t.Helper()
				if request.GetRequestUnit() == nil {
					t.Fatal("legacy request_unit did not decode")
				}
			},
		},
		{
			name: "topology spec remains field 20",
			wire: []byte{0xa2, 0x01, 0x00},
			validate: func(t *testing.T, request *clusterngv1beta.CreateSharedPoolRequest) {
				t.Helper()
				if request.GetTopologySpec() == nil {
					t.Fatal("legacy topology_spec did not decode")
				}
			},
		},
		{
			name: "new fields remain 23 through 25",
			wire: []byte{0xb8, 0x01, 0x01, 0xc0, 0x01, 0x01, 0xca, 0x01, 0x00},
			validate: func(t *testing.T, request *clusterngv1beta.CreateSharedPoolRequest) {
				t.Helper()
				if request.GetTiflashMode() == 0 || !request.GetEnforceO11YNetworkNonOverlap() || request.GetBootstrapCapacity() == nil {
					t.Fatalf("new CSNG creation fields did not decode: %v", request)
				}
			},
		},
	}

	for _, test := range tests {
		t.Run(test.name, func(t *testing.T) {
			t.Parallel()

			request := &clusterngv1beta.CreateSharedPoolRequest{}
			if err := proto.Unmarshal(test.wire, request); err != nil {
				t.Fatalf("unmarshal wire contract: %v", err)
			}
			test.validate(t, request)
		})
	}
}

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

	optionFields := (&clusterngv1beta.ExportOptions{}).ProtoReflect().Descriptor().Fields()
	escapeBackslash := optionFields.ByName("escape_backslash")
	if escapeBackslash == nil || escapeBackslash.Number() != 5 || escapeBackslash.Kind() != protoreflect.BoolKind ||
		!escapeBackslash.HasPresence() {
		t.Errorf("escape_backslash descriptor = %v, want optional bool field 5 with presence", escapeBackslash)
	}
	parquetFormat := optionFields.ByName("parquet_format")
	if parquetFormat == nil || parquetFormat.Number() != 6 || parquetFormat.Kind() != protoreflect.MessageKind ||
		!parquetFormat.HasPresence() {
		t.Errorf("parquet_format descriptor = %v, want optional message field 6 with presence", parquetFormat)
	}

	tableFields := (&clusterngv1beta.ExportOptions_Filter_Table{}).ProtoReflect().Descriptor().Fields()
	partitions := tableFields.ByName("partitions")
	if partitions == nil || partitions.Number() != 3 || partitions.Kind() != protoreflect.StringKind ||
		partitions.Cardinality() != protoreflect.Repeated || partitions.HasPresence() {
		t.Errorf("partitions descriptor = %v, want repeated string field 3 without presence", partitions)
	}

	if clusterngv1beta.ExportOptions_PARQUET != 3 {
		t.Errorf("ExportOptions.PARQUET = %d, want 3", clusterngv1beta.ExportOptions_PARQUET)
	}
	for compression, number := range map[clusterngv1beta.ExportParquetCompressionType_Enum]int32{
		clusterngv1beta.ExportParquetCompressionType_GZIP:   1,
		clusterngv1beta.ExportParquetCompressionType_SNAPPY: 2,
		clusterngv1beta.ExportParquetCompressionType_ZSTD:   3,
		clusterngv1beta.ExportParquetCompressionType_NONE:   4,
	} {
		if int32(compression) != number {
			t.Errorf("ExportParquetCompressionType.%s = %d, want %d", compression, compression, number)
		}
	}
}

func assertCSNGMethod(t *testing.T, serviceName, methodName protoreflect.Name) protoreflect.MethodDescriptor {
	t.Helper()

	service := clusterngv1beta.File_cluster_service_ng_v1beta_services_proto.Services().ByName(serviceName)
	if service == nil {
		t.Fatalf("CSNG method %s.%s is not available", serviceName, methodName)
	}
	method := service.Methods().ByName(methodName)
	if method == nil {
		t.Fatalf("CSNG method %s.%s is not available", serviceName, methodName)
	}
	return method
}