package common

import (
	"testing"

	"github.com/go-logr/logr"
	"github.com/stretchr/testify/assert"
	"github.com/stretchr/testify/require"
	"k8s.io/apimachinery/pkg/runtime"
	"sigs.k8s.io/controller-runtime/pkg/client/fake"
	"sigs.k8s.io/yaml"

	"github.com/tidbcloud/infra-provider/regional-server/api/v1alpha1"
	"github.com/tidbcloud/infra-provider/regional-server/pkg/config"
	"github.com/tidbcloud/infra-provider/regional-server/pkg/controllers/privatelinkservice/task"
	"github.com/tidbcloud/infra-provider/regional-server/pkg/k8s"
	"github.com/tidbcloud/infra-provider/regional-server/pkg/namer"
	"github.com/tidbcloud/infra-provider/regional-server/pkg/repository"
	"github.com/tidbcloud/infra-provider/regional-server/test/testdata"
)

func fakeTaskContext(t *testing.T) *task.Context {
	s := runtime.NewScheme()
	v1alpha1.AddToScheme(s)
	fakeVPC := &v1alpha1.VPC{}
	bs, err := testdata.LoadDefaultVPCYAML()
	require.NoError(t, err)
	require.NoError(t, yaml.Unmarshal(bs, fakeVPC))
	fakepls := &v1alpha1.PrivateLinkService{}
	bs, err = testdata.LoadEKSPrivateLinkService()
	require.NoError(t, err)
	assert.NoError(t, yaml.Unmarshal(bs, fakepls))
	cli := fake.NewClientBuilder().WithScheme(s).WithObjects(fakeVPC, fakepls).WithStatusSubresource(fakeVPC, fakepls).Build()
	c := &task.Context{
		Local:                k8s.NewMutClientWithOwner(cli, fakeVPC.DeepCopy(), namer.NewNamer()).Client(),
		Logger:               logr.Discard(),
		VPC:                  fakeVPC.DeepCopy(),
		VPCR:                 repository.NewVPCReader(cli),
		Config:               config.RootConfig{},
		PrivateLinkService:   fakepls.DeepCopy(),
		PrivateLinkServiceRW: repository.NewPrivateLinkServiceReaderWriter(cli),
	}
	c, err = c.Default()
	assert.Nil(t, err)
	return c
}

func TestPrimaryAndListenerPorts(t *testing.T) {
	cases := []struct {
		name          string
		spec          v1alpha1.LoadBalancerSpec
		wantPrimary   int32
		wantListeners []int32
	}{
		{
			name:          "legacy single port only",
			spec:          v1alpha1.LoadBalancerSpec{Port: 4000},
			wantPrimary:   4000,
			wantListeners: nil,
		},
		{
			name: "multi-port populates listeners and picks first as primary",
			spec: v1alpha1.LoadBalancerSpec{Ports: []v1alpha1.LoadBalancerPort{
				{Name: "cassandra", Port: 9042},
				{Name: "cassandra-ssl", Port: 9142},
			}},
			wantPrimary:   9042,
			wantListeners: []int32{9042, 9142},
		},
		{
			name: "ports with zero entries collapses to legacy port",
			spec: v1alpha1.LoadBalancerSpec{
				Port:  4000,
				Ports: []v1alpha1.LoadBalancerPort{{Port: 0}},
			},
			wantPrimary:   4000,
			wantListeners: nil,
		},
		{
			name:          "neither set returns zero primary",
			spec:          v1alpha1.LoadBalancerSpec{},
			wantPrimary:   0,
			wantListeners: nil,
		},
	}

	for _, tc := range cases {
		t.Run(tc.name, func(t *testing.T) {
			lb := &v1alpha1.LoadBalancer{Spec: tc.spec}
			gotPrimary, gotListeners := primaryAndListenerPorts(lb)
			assert.Equal(t, tc.wantPrimary, gotPrimary)
			assert.Equal(t, tc.wantListeners, gotListeners)
		})
	}
}