package service

import (
	"context"
	"crypto/rand"
	"crypto/rsa"
	"crypto/x509"
	"encoding/base64"
	"encoding/pem"
	"fmt"
	"net/http"
	"testing"

	"github.com/golang/mock/gomock"
	tfilter "github.com/pingcap/tidb/pkg/util/table-filter"
	"github.com/stretchr/testify/assert"
	"github.com/stretchr/testify/require"
	apiv1 "github.com/tidbcloud/cluster-service-ng/api/tidbcloud/ticdc-service/v1"
	"github.com/tidbcloud/service-proxy/pkg/checker/core"
	"google.golang.org/grpc/codes"
	"google.golang.org/grpc/status"

	"github.com/tidbcloud/ticdc-service-ng/pkg/util"

	"github.com/tidbcloud/ticdc-service-ng/pkg/clusteragent"
	agentMock "github.com/tidbcloud/ticdc-service-ng/pkg/clusteragent/mocks"
	"github.com/tidbcloud/ticdc-service-ng/pkg/clusterproxy"
	"github.com/tidbcloud/ticdc-service-ng/pkg/clusterproxy/api"
	apiMock "github.com/tidbcloud/ticdc-service-ng/pkg/clusterproxy/api/mocks"
	proxyModelMock "github.com/tidbcloud/ticdc-service-ng/pkg/clusterproxy/mocks"
)

func generateTestPEM(t *testing.T) []byte {
	t.Helper()
	privateKey, err := rsa.GenerateKey(rand.Reader, 2048)
	require.NoError(t, err)

	pubASN1, err := x509.MarshalPKIXPublicKey(&privateKey.PublicKey)
	require.NoError(t, err)

	return pem.EncodeToMemory(&pem.Block{
		Type:  "PUBLIC KEY",
		Bytes: pubASN1,
	})
}

func mustNewPreCheckServiceServer(
	t *testing.T,
	proxyModel clusterproxy.Model,
	opts ...PreCheckServerOption,
) apiv1.TiCDCPrecheckServiceServer {
	t.Helper()

	server, err := NewPreCheckServiceServer(proxyModel, opts...)
	require.NoError(t, err)
	return server
}

func TestPreCheckServerImpl_TestNetworkConnection(t *testing.T) {
	ctrl := gomock.NewController(t)
	defer ctrl.Finish()

	mockProxy := proxyModelMock.NewMockClusterProxy(ctrl)
	modelMock := proxyModelMock.NewMockModel(ctrl)
	server := mustNewPreCheckServiceServer(t, modelMock)

	req := &apiv1.TestNetworkConnectionRequest{
		ClusterId: "test-cluster",
		Type:      "mysql",
		MysqlParam: &apiv1.TestNetworkConnectionRequest_MySQLParameter{
			User:     "user",
			Password: "password",
			Url:      "url",
		},
	}

	modelMock.EXPECT().GetClusterProxy(gomock.Any(), req.ClusterId, 300).Return(mockProxy, nil)
	cfCli := apiMock.NewMockChangefeedAPIV2(ctrl)
	mockProxy.EXPECT().ChangefeedClient().Return(cfCli)
	cfCli.EXPECT().TestConnection(gomock.Any(), gomock.Any()).Return(nil)

	_, err := server.TestNetworkConnection(context.Background(), req)
	assert.NoError(t, err)
}

func TestPreCheckServerImpl_ValidateChangefeedParameters(t *testing.T) {
	ctrl := gomock.NewController(t)
	defer ctrl.Finish()

	mockProxy := proxyModelMock.NewMockClusterProxy(ctrl)
	modelMock := proxyModelMock.NewMockModel(ctrl)
	server := mustNewPreCheckServiceServer(t, modelMock)

	req := &apiv1.ValidateChangefeedParametersRequest{
		ClusterId: "test-cluster",
		Type:      "kafka",
		KafkaParam: &apiv1.ValidateChangefeedParametersRequest_KafkaParameter{
			User:      "user",
			Password:  "password",
			Url:       "url",
			EnableTls: true,
			AuthType:  "sasl-plain",
			Topic:     "topic",
		},
	}

	modelMock.EXPECT().GetClusterProxy(gomock.Any(), req.ClusterId, 300).Return(mockProxy, nil)
	cfCli := apiMock.NewMockChangefeedAPIV2(ctrl)
	mockProxy.EXPECT().ChangefeedClient().Return(cfCli)
	cfCli.EXPECT().TestConnection(gomock.Any(), gomock.Any()).Return(nil)

	_, err := server.ValidateChangefeedParameters(context.Background(), req)
	assert.NoError(t, err)
}

func TestPreCheckServerImpl_VerifyTable(t *testing.T) {
	ctrl := gomock.NewController(t)
	defer ctrl.Finish()

	mockProxy := proxyModelMock.NewMockClusterProxy(ctrl)
	modelMock := proxyModelMock.NewMockModel(ctrl)
	mockAgentModel := agentMock.NewMockModel(ctrl)
	mockAgentClient := agentMock.NewMockClient(ctrl)

	server := mustNewPreCheckServiceServer(t, modelMock, WithClusterAgentModel(mockAgentModel))

	testPEM := generateTestPEM(t)

	req := &apiv1.VerifyTableRequest{
		TidbClusterId: "test-cluster",
		User:          "user",
		Password:      "password",
		FilterRules:   []string{"*.*"},
	}

	mockAgentModel.EXPECT().GetClient(gomock.Any(), req.TidbClusterId, 300).Return(mockAgentClient, nil)
	mockAgentClient.EXPECT().GetPublicKey(gomock.Any()).Return(testPEM, nil)
	// Return tables including system schemas that should be filtered out
	mockAgentClient.EXPECT().QueryTiCDCTablesEligibility(gomock.Any(), req.TidbClusterId, gomock.Any(), gomock.Any()).
		Return(
			&clusteragent.TablesEligibilityResult{
				Schemas: []string{"test", "tidb_cdc", "lightning_task_info"},
				Tables:  []string{"test", "tidb_cdc", "lightning_task_info"},
			},
			&clusteragent.TablesEligibilityResult{
				Schemas: []string{"test", "tidb_cdc", "lightning_task_info"},
				Tables:  []string{"test", "tidb_cdc", "lightning_task_info"},
			},
			nil,
		)

	modelMock.EXPECT().GetClusterProxy(gomock.Any(), req.TidbClusterId, 300).Return(mockProxy, nil)
	pdCli := apiMock.NewMockPDClient(ctrl)
	mockProxy.EXPECT().PDClient().Return(pdCli)
	pdCli.EXPECT().GetPDSafepoint(gomock.Any()).Return(&api.Safepoint{
		SP: uint64(10),
	}, nil)

	resp, err := server.VerifyTable(context.Background(), req)
	assert.NoError(t, err)
	assert.NotNil(t, resp)
	assert.Equal(t, 1, len(resp.EligibleTables))
	assert.Equal(t, "test.test", resp.EligibleTables[0])
	assert.Equal(t, 1, len(resp.IneligibleTables))
	assert.Equal(t, "test.test", resp.IneligibleTables[0])
	assert.Equal(t, uint64(10), resp.GcSafePoint)
}

func TestPreCheckServerImpl_VerifyTable_InvalidFilterRules(t *testing.T) {
	ctrl := gomock.NewController(t)
	defer ctrl.Finish()

	modelMock := proxyModelMock.NewMockModel(ctrl)
	server := mustNewPreCheckServiceServer(t, modelMock)
	req := &apiv1.VerifyTableRequest{
		TidbClusterId: "test-cluster",
		User:          "user",
		Password:      "password",
		FilterRules:   []string{"["},
		CaseSensitive: false,
	}

	var (
		resp *apiv1.VerifyTableResponse
		err  error
	)
	require.NotPanics(t, func() {
		resp, err = server.VerifyTable(context.Background(), req)
	})
	require.Nil(t, resp)
	require.Error(t, err)

	st, ok := status.FromError(err)
	require.True(t, ok)
	assert.Equal(t, codes.InvalidArgument, st.Code())
}

func TestNewPreCheckServiceServer_ProxyModelNil(t *testing.T) {
	server, err := NewPreCheckServiceServer(nil)
	require.Nil(t, server)
	require.Error(t, err)
}

func TestPreCheckServerImpl_VerifyTable_GetClientError(t *testing.T) {
	ctrl := gomock.NewController(t)
	defer ctrl.Finish()

	modelMock := proxyModelMock.NewMockModel(ctrl)
	mockAgentModel := agentMock.NewMockModel(ctrl)

	server := mustNewPreCheckServiceServer(t, modelMock, WithClusterAgentModel(mockAgentModel))

	req := &apiv1.VerifyTableRequest{
		TidbClusterId: "test-cluster",
		User:          "user",
		Password:      "password",
	}

	mockAgentModel.EXPECT().GetClient(gomock.Any(), req.TidbClusterId, 300).
		Return(nil, fmt.Errorf("connection failed"))

	_, err := server.VerifyTable(context.Background(), req)
	require.Error(t, err)

	st, ok := status.FromError(err)
	require.True(t, ok)
	assert.Equal(t, codes.Internal, st.Code())
}

func TestPreCheckServerImpl_VerifyTable_AgentPath_PreserveStatusCodeFromGetClient(t *testing.T) {
	ctrl := gomock.NewController(t)
	defer ctrl.Finish()

	modelMock := proxyModelMock.NewMockModel(ctrl)
	mockAgentModel := agentMock.NewMockModel(ctrl)

	server := mustNewPreCheckServiceServer(t, modelMock, WithClusterAgentModel(mockAgentModel))

	req := &apiv1.VerifyTableRequest{
		TidbClusterId: "test-cluster",
		User:          "user",
		Password:      "password",
	}

	mockAgentModel.EXPECT().GetClient(gomock.Any(), req.TidbClusterId, 300).
		Return(nil, status.Error(codes.NotFound, "tidb group not found"))

	_, err := server.VerifyTable(context.Background(), req)
	require.Error(t, err)

	st, ok := status.FromError(err)
	require.True(t, ok)
	assert.Equal(t, codes.NotFound, st.Code())
}

func TestNewPreCheckServiceServer_WithClusterAgentModelNil(t *testing.T) {
	ctrl := gomock.NewController(t)
	defer ctrl.Finish()

	modelMock := proxyModelMock.NewMockModel(ctrl)
	server, err := NewPreCheckServiceServer(modelMock, WithClusterAgentModel(nil))
	require.Nil(t, server)
	require.Error(t, err)
}

func TestPreCheckServerImpl_VerifyTable_GetPublicKeyError(t *testing.T) {
	ctrl := gomock.NewController(t)
	defer ctrl.Finish()

	modelMock := proxyModelMock.NewMockModel(ctrl)
	mockAgentModel := agentMock.NewMockModel(ctrl)
	mockAgentClient := agentMock.NewMockClient(ctrl)

	server := mustNewPreCheckServiceServer(t, modelMock, WithClusterAgentModel(mockAgentModel))

	req := &apiv1.VerifyTableRequest{
		TidbClusterId: "test-cluster",
		User:          "user",
		Password:      "password",
	}

	mockAgentModel.EXPECT().GetClient(gomock.Any(), req.TidbClusterId, 300).Return(mockAgentClient, nil)
	mockAgentClient.EXPECT().GetPublicKey(gomock.Any()).Return(nil, fmt.Errorf("key not available"))

	_, err := server.VerifyTable(context.Background(), req)
	require.Error(t, err)

	st, ok := status.FromError(err)
	require.True(t, ok)
	assert.Equal(t, codes.Internal, st.Code())
}

func TestPreCheckServerImpl_VerifyTable_QueryError(t *testing.T) {
	ctrl := gomock.NewController(t)
	defer ctrl.Finish()

	modelMock := proxyModelMock.NewMockModel(ctrl)
	mockAgentModel := agentMock.NewMockModel(ctrl)
	mockAgentClient := agentMock.NewMockClient(ctrl)

	server := mustNewPreCheckServiceServer(t, modelMock, WithClusterAgentModel(mockAgentModel))

	testPEM := generateTestPEM(t)

	req := &apiv1.VerifyTableRequest{
		TidbClusterId: "test-cluster",
		User:          "user",
		Password:      "password",
	}

	mockAgentModel.EXPECT().GetClient(gomock.Any(), req.TidbClusterId, 300).Return(mockAgentClient, nil)
	mockAgentClient.EXPECT().GetPublicKey(gomock.Any()).Return(testPEM, nil)
	mockAgentClient.EXPECT().QueryTiCDCTablesEligibility(gomock.Any(), req.TidbClusterId, gomock.Any(), gomock.Any()).
		Return(nil, nil, fmt.Errorf("query failed"))
	modelMock.EXPECT().GetClusterProxy(gomock.Any(), req.TidbClusterId, 300).
		AnyTimes().Return(nil, nil)

	_, err := server.VerifyTable(context.Background(), req)
	require.Error(t, err)

	st, ok := status.FromError(err)
	require.True(t, ok)
	assert.Equal(t, codes.Internal, st.Code())
}

func TestPreCheckServerImpl_VerifyTable_AgentPath_HTTPStatusErrorMapping(t *testing.T) {
	tests := []struct {
		name      string
		setupMock func(
			t *testing.T,
			modelMock *proxyModelMock.MockModel,
			agentModel *agentMock.MockModel,
			agentClient *agentMock.MockClient,
			req *apiv1.VerifyTableRequest,
		)
		wantCode codes.Code
	}{
		{
			name: "get public key 401 maps to Unauthenticated",
			setupMock: func(
				t *testing.T,
				modelMock *proxyModelMock.MockModel,
				agentModel *agentMock.MockModel,
				agentClient *agentMock.MockClient,
				req *apiv1.VerifyTableRequest,
			) {
				agentModel.EXPECT().GetClient(gomock.Any(), req.TidbClusterId, 300).Return(agentClient, nil)
				agentClient.EXPECT().GetPublicKey(gomock.Any()).
					Return(nil, &util.StatusError{StatusCode: http.StatusUnauthorized})
			},
			wantCode: codes.Unauthenticated,
		},
		{
			name: "query 500 maps to Unavailable",
			setupMock: func(
				t *testing.T,
				modelMock *proxyModelMock.MockModel,
				agentModel *agentMock.MockModel,
				agentClient *agentMock.MockClient,
				req *apiv1.VerifyTableRequest,
			) {
				testPEM := generateTestPEM(t)
				agentModel.EXPECT().GetClient(gomock.Any(), req.TidbClusterId, 300).Return(agentClient, nil)
				agentClient.EXPECT().GetPublicKey(gomock.Any()).Return(testPEM, nil)
				agentClient.EXPECT().QueryTiCDCTablesEligibility(gomock.Any(), req.TidbClusterId, gomock.Any(), gomock.Any()).
					Return(nil, nil, &util.StatusError{StatusCode: http.StatusInternalServerError})
				modelMock.EXPECT().GetClusterProxy(gomock.Any(), req.TidbClusterId, 300).
					AnyTimes().Return(nil, nil)
			},
			wantCode: codes.Unavailable,
		},
		{
			name: "query 408 maps to DeadlineExceeded",
			setupMock: func(
				t *testing.T,
				modelMock *proxyModelMock.MockModel,
				agentModel *agentMock.MockModel,
				agentClient *agentMock.MockClient,
				req *apiv1.VerifyTableRequest,
			) {
				testPEM := generateTestPEM(t)
				agentModel.EXPECT().GetClient(gomock.Any(), req.TidbClusterId, 300).Return(agentClient, nil)
				agentClient.EXPECT().GetPublicKey(gomock.Any()).Return(testPEM, nil)
				agentClient.EXPECT().QueryTiCDCTablesEligibility(gomock.Any(), req.TidbClusterId, gomock.Any(), gomock.Any()).
					Return(nil, nil, &util.StatusError{StatusCode: http.StatusRequestTimeout})
				modelMock.EXPECT().GetClusterProxy(gomock.Any(), req.TidbClusterId, 300).
					AnyTimes().Return(nil, nil)
			},
			wantCode: codes.DeadlineExceeded,
		},
	}

	for _, tt := range tests {
		t.Run(tt.name, func(t *testing.T) {
			ctrl := gomock.NewController(t)
			defer ctrl.Finish()

			modelMock := proxyModelMock.NewMockModel(ctrl)
			mockAgentModel := agentMock.NewMockModel(ctrl)
			mockAgentClient := agentMock.NewMockClient(ctrl)
			server := mustNewPreCheckServiceServer(t, modelMock, WithClusterAgentModel(mockAgentModel))

			req := &apiv1.VerifyTableRequest{
				TidbClusterId: "test-cluster",
				User:          "user",
				Password:      "password",
			}

			tt.setupMock(t, modelMock, mockAgentModel, mockAgentClient, req)

			_, err := server.VerifyTable(context.Background(), req)
			require.Error(t, err)

			st, ok := status.FromError(err)
			require.True(t, ok)
			assert.Equal(t, tt.wantCode, st.Code())
		})
	}
}

func TestBuildCheckRequest(t *testing.T) {
	tests := []struct {
		name    string
		req     *apiv1.TestNetworkConnectionRequest
		want    *core.CheckRequest
		wantErr bool
	}{
		{
			name: "MySQL parameters",
			req: &apiv1.TestNetworkConnectionRequest{
				Type: "mysql",
				MysqlParam: &apiv1.TestNetworkConnectionRequest_MySQLParameter{
					User:     "user",
					Password: "password",
					Url:      "host:3307",
				},
			},
			want: &core.CheckRequest{
				TargetType: "mysql",
				MySQL: &core.MySQLParams{
					Host:     "host",
					Port:     3307,
					User:     "user",
					Password: "password",
				},
			},
			wantErr: false,
		},
		{
			name: "MySQL parameters with default port",
			req: &apiv1.TestNetworkConnectionRequest{
				Type: "mysql",
				MysqlParam: &apiv1.TestNetworkConnectionRequest_MySQLParameter{
					User:     "user",
					Password: "password",
					Url:      "host",
				},
			},
			want: &core.CheckRequest{
				TargetType: "mysql",
				MySQL: &core.MySQLParams{
					Host:     "host",
					Port:     3306,
					User:     "user",
					Password: "password",
				},
			},
			wantErr: false,
		},
		{
			name: "Kafka parameters",
			req: &apiv1.TestNetworkConnectionRequest{
				Type: "kafka",
				KafkaParam: &apiv1.TestNetworkConnectionRequest_KafkaParameter{
					User:              "user",
					Password:          "password",
					EnableTls:         true,
					AuthType:          "sasl-plain",
					BoostrapEndpoints: "kafka://127.0.0.1:9092",
				},
			},
			want: &core.CheckRequest{
				TargetType: "kafka",
				Kafka: &core.KafkaParams{
					Brokers:   "kafka://127.0.0.1:9092",
					User:      "user",
					Password:  "password",
					EnableTLS: true,
					AuthType:  "sasl-plain",
				},
			},
			wantErr: false,
		},
		{
			name: "S3 parameters",
			req: &apiv1.TestNetworkConnectionRequest{
				Type: "s3",
				S3Param: &apiv1.TestNetworkConnectionRequest_S3Parameter{
					S3Uri:           "s3://bucket",
					AccessKeyId:     "accessKeyId",
					SecretAccessKey: "secretAccessKey",
					RoleArn:         "roleArn",
					ExternalId:      "externalId",
				},
			},
			want: &core.CheckRequest{
				TargetType: "s3",
				S3: &core.S3Params{
					URI: "s3://bucket",
					AWSAuth: core.AWSAuth{
						AccessKeyID:     "accessKeyId",
						SecretAccessKey: "secretAccessKey",
						RoleARN:         "roleArn",
						ExternalID:      "externalId",
					},
				},
			},
			wantErr: false,
		},
		{
			name: "S3 parameters with custom endpoint",
			req: &apiv1.TestNetworkConnectionRequest{
				Type: "s3",
				S3Param: &apiv1.TestNetworkConnectionRequest_S3Parameter{
					S3Uri:           "s3://bucket",
					AccessKeyId:     "accessKeyId",
					SecretAccessKey: "secretAccessKey",
					Endpoint:        "https://s3.oss-us-east-1.aliyuncs.com",
				},
			},
			want: &core.CheckRequest{
				TargetType: "s3",
				S3: &core.S3Params{
					URI:      "s3://bucket",
					Endpoint: "https://s3.oss-us-east-1.aliyuncs.com",
					AWSAuth: core.AWSAuth{
						AccessKeyID:     "accessKeyId",
						SecretAccessKey: "secretAccessKey",
					},
				},
			},
			wantErr: false,
		},
		{
			name: "GCS parameters",
			req: &apiv1.TestNetworkConnectionRequest{
				Type: "gcs",
				GcsParam: &apiv1.TestNetworkConnectionRequest_GCSParameter{
					GcsUrl:                    "gcs://bucket",
					Region:                    "region",
					ImpersonateServiceAccount: "serviceAccount",
				},
			},
			want: &core.CheckRequest{
				TargetType: "gcs",
				GCS: &core.GCSParams{
					Url:                       "gcs://bucket",
					Region:                    "region",
					ImpersonateServiceAccount: "serviceAccount",
				},
			},
			wantErr: false,
		},
		{
			name: "pulsar parameters",
			req: &apiv1.TestNetworkConnectionRequest{
				Type: "pulsar",
				PulsarParam: &apiv1.TestNetworkConnectionRequest_PulsarParameter{
					ServiceUrl:         "pulsar://pulsar-service",
					AuthenticationType: "basic",
					Username:           "username",
					Password:           "password",
					Token:              "",
					ConnectionTimeout:  5,
					Tenant:             "public",
					Namespace:          "default",
					Topic:              "test",
				},
			},
			want: &core.CheckRequest{
				TargetType: "pulsar",
				Pulsar: &core.PulsarParams{
					ServiceURL:        "pulsar://pulsar-service",
					AuthType:          "basic",
					Username:          "username",
					Password:          "password",
					Token:             "",
					ConnectionTimeout: 5,
					Tenant:            "public",
					Namespace:         "default",
					Topic:             "test",
				},
			},
			wantErr: false,
		},
		{
			name:    "nil request",
			req:     nil,
			want:    nil,
			wantErr: true,
		},
		{
			name: "mysql with nil params",
			req: &apiv1.TestNetworkConnectionRequest{
				Type:       "mysql",
				MysqlParam: nil,
			},
			want:    nil,
			wantErr: true,
		},
		{
			name: "kafka with nil params",
			req: &apiv1.TestNetworkConnectionRequest{
				Type:       "kafka",
				KafkaParam: nil,
			},
			want:    nil,
			wantErr: true,
		},
		{
			name: "s3 with nil params",
			req: &apiv1.TestNetworkConnectionRequest{
				Type:    "s3",
				S3Param: nil,
			},
			want:    nil,
			wantErr: true,
		},
		{
			name: "gcs with nil params",
			req: &apiv1.TestNetworkConnectionRequest{
				Type:     "gcs",
				GcsParam: nil,
			},
			want:    nil,
			wantErr: true,
		},
		{
			name: "pulsar with nil params",
			req: &apiv1.TestNetworkConnectionRequest{
				Type:        "pulsar",
				PulsarParam: nil,
			},
			want:    nil,
			wantErr: true,
		},
		{
			name: "unsupported type",
			req: &apiv1.TestNetworkConnectionRequest{
				Type: "unknown",
			},
			want:    nil,
			wantErr: true,
		},
		{
			name: "kafka with invalid auth type",
			req: &apiv1.TestNetworkConnectionRequest{
				Type: "kafka",
				KafkaParam: &apiv1.TestNetworkConnectionRequest_KafkaParameter{
					User:              "user",
					Password:          "password",
					EnableTls:         true,
					AuthType:          "invalid-auth-type",
					BoostrapEndpoints: "kafka://127.0.0.1:9092",
				},
			},
			want:    nil,
			wantErr: true,
		},
		{
			name: "pulsar with invalid auth type",
			req: &apiv1.TestNetworkConnectionRequest{
				Type: "pulsar",
				PulsarParam: &apiv1.TestNetworkConnectionRequest_PulsarParameter{
					ServiceUrl:         "pulsar://pulsar-service",
					AuthenticationType: "oauth2",
					Username:           "username",
					Password:           "password",
				},
			},
			want:    nil,
			wantErr: true,
		},
	}

	for _, tt := range tests {
		t.Run(tt.name, func(t *testing.T) {
			got, err := buildCheckRequest(tt.req)
			if tt.wantErr {
				assert.Error(t, err)
				assert.Nil(t, got)
			} else {
				assert.NoError(t, err)
				assert.Equal(t, tt.want, got)
			}
		})
	}
}

func TestParseHostPort(t *testing.T) {
	tests := []struct {
		name     string
		addr     string
		wantHost string
		wantPort int
		wantErr  bool
	}{
		{"simple host:port", "localhost:3306", "localhost", 3306, false},
		{"custom port", "localhost:3307", "localhost", 3307, false},
		{"no port", "localhost", "localhost", 3306, false},
		{"ipv4 with port", "192.168.1.1:3306", "192.168.1.1", 3306, false},
		{"ipv6 with brackets and port", "[::1]:3306", "::1", 3306, false},
		{"ipv6 without port", "[::1]", "[::1]", 3306, false},
		{"invalid port string", "localhost:abc", "localhost", 0, true},
		{"port out of range high", "localhost:99999", "localhost", 0, true},
		{"port out of range zero", "localhost:0", "localhost", 0, true},
		{"empty string", "", "", 3306, false},
		{"hostname with subdomain", "db.example.com:3306", "db.example.com", 3306, false},
	}

	for _, tt := range tests {
		t.Run(tt.name, func(t *testing.T) {
			host, port, err := parseHostPort(tt.addr)
			if tt.wantErr {
				assert.Error(t, err)
			} else {
				assert.NoError(t, err)
			}
			assert.Equal(t, tt.wantHost, host)
			assert.Equal(t, tt.wantPort, port)
		})
	}
}

func TestBuildKafkaCheckRequest(t *testing.T) {
	tests := []struct {
		name    string
		req     *apiv1.ValidateChangefeedParametersRequest
		want    *core.CheckRequest
		wantErr bool
	}{
		{
			name: "valid kafka parameters",
			req: &apiv1.ValidateChangefeedParametersRequest{
				Type: "kafka",
				KafkaParam: &apiv1.ValidateChangefeedParametersRequest_KafkaParameter{
					Url:                     "localhost:9092",
					User:                    "user",
					Password:                "password",
					EnableTls:               true,
					AuthType:                "sasl-scram-sha-256",
					ReplicationFactor:       3,
					PartitionNum:            1,
					SchemaRegistryEndpoints: "http://registry:8081",
				},
			},
			want: &core.CheckRequest{
				TargetType: "kafka",
				Kafka: &core.KafkaParams{
					Brokers:                 "localhost:9092",
					User:                    "user",
					Password:                "password",
					EnableTLS:               true,
					AuthType:                "sasl-scram-sha-256",
					ReplicationFactor:       3,
					PartitionNum:            1,
					SchemaRegistryEndpoints: "http://registry:8081",
				},
			},
			wantErr: false,
		},
		{
			name:    "nil request",
			req:     nil,
			want:    nil,
			wantErr: true,
		},
		{
			name: "nil kafka params",
			req: &apiv1.ValidateChangefeedParametersRequest{
				Type:       "kafka",
				KafkaParam: nil,
			},
			want:    nil,
			wantErr: true,
		},
		{
			name: "invalid kafka auth type",
			req: &apiv1.ValidateChangefeedParametersRequest{
				Type: "kafka",
				KafkaParam: &apiv1.ValidateChangefeedParametersRequest_KafkaParameter{
					Url:      "localhost:9092",
					AuthType: "invalid-auth",
				},
			},
			want:    nil,
			wantErr: true,
		},
	}

	for _, tt := range tests {
		t.Run(tt.name, func(t *testing.T) {
			got, err := buildKafkaCheckRequest(tt.req)
			if tt.wantErr {
				assert.Error(t, err)
				assert.Nil(t, got)
			} else {
				assert.NoError(t, err)
				assert.Equal(t, tt.want, got)
			}
		})
	}
}

func TestPreCheckServerImpl_VerifyTable_LegacyPath(t *testing.T) {
	ctrl := gomock.NewController(t)
	defer ctrl.Finish()

	mockProxy := proxyModelMock.NewMockClusterProxy(ctrl)
	modelMock := proxyModelMock.NewMockModel(ctrl)

	// Use default options to force clusterproxy SQL path.
	server := mustNewPreCheckServiceServer(t, modelMock)

	req := &apiv1.VerifyTableRequest{
		TidbClusterId: "test-cluster",
		User:          "user",
		Password:      "password",
		FilterRules:   []string{"*.*"},
	}

	modelMock.EXPECT().GetClusterProxy(gomock.Any(), req.TidbClusterId, 300).Return(mockProxy, nil)

	sqlCli := apiMock.NewMockSQLClient(ctrl)
	mockProxy.EXPECT().SQLClient().Return(sqlCli).Times(2)

	// eligible query: return base64-encoded schema/table names
	sqlCli.EXPECT().QuerySQL(gomock.Any(), gomock.Any(), QueryEligibleSchemaTableClaus, gomock.Any()).
		DoAndReturn(func(_ context.Context, _ *api.ClusterOperator, _ string, dest interface{}) error {
			info := dest.(*TableSchemaInfo)
			info.TableSchema = []string{
				base64.StdEncoding.EncodeToString([]byte("test_db")),
				base64.StdEncoding.EncodeToString([]byte("tidb_cdc")),
			}
			info.TableName = []string{
				base64.StdEncoding.EncodeToString([]byte("users")),
				base64.StdEncoding.EncodeToString([]byte("cdc_table")),
			}
			return nil
		})

	// ineligible query
	sqlCli.EXPECT().QuerySQL(gomock.Any(), gomock.Any(), QueryIneligibleSchemaTableClaus, gomock.Any()).
		DoAndReturn(func(_ context.Context, _ *api.ClusterOperator, _ string, dest interface{}) error {
			info := dest.(*TableSchemaInfo)
			info.TableSchema = []string{
				base64.StdEncoding.EncodeToString([]byte("test_db")),
			}
			info.TableName = []string{
				base64.StdEncoding.EncodeToString([]byte("no_pk_table")),
			}
			return nil
		})

	pdCli := apiMock.NewMockPDClient(ctrl)
	mockProxy.EXPECT().PDClient().Return(pdCli)
	pdCli.EXPECT().GetPDSafepoint(gomock.Any()).Return(&api.Safepoint{SP: uint64(42)}, nil)

	resp, err := server.VerifyTable(context.Background(), req)
	require.NoError(t, err)
	require.NotNil(t, resp)
	// tidb_cdc should be filtered out as system schema
	assert.Equal(t, []string{"test_db.users"}, resp.EligibleTables)
	assert.Equal(t, []string{"test_db.no_pk_table"}, resp.IneligibleTables)
	assert.Equal(t, uint64(42), resp.GcSafePoint)
}

func TestPreCheckServerImpl_VerifyTable_LegacyPath_ProxyError(t *testing.T) {
	ctrl := gomock.NewController(t)
	defer ctrl.Finish()

	modelMock := proxyModelMock.NewMockModel(ctrl)
	server := mustNewPreCheckServiceServer(t, modelMock)

	req := &apiv1.VerifyTableRequest{
		TidbClusterId: "test-cluster",
		User:          "user",
		Password:      "password",
	}

	modelMock.EXPECT().GetClusterProxy(gomock.Any(), req.TidbClusterId, 300).
		Return(nil, fmt.Errorf("proxy unavailable"))

	_, err := server.VerifyTable(context.Background(), req)
	require.Error(t, err)

	st, ok := status.FromError(err)
	require.True(t, ok)
	assert.Equal(t, codes.Internal, st.Code())
}

func TestParseQueryTable(t *testing.T) {
	// Use a wildcard filter that matches everything
	filter, err := tfilter.Parse([]string{"*.*"})
	require.NoError(t, err)

	tests := []struct {
		name    string
		info    *TableSchemaInfo
		want    []string
		wantErr bool
	}{
		{
			name: "valid base64 with system schema filtered",
			info: &TableSchemaInfo{
				TableSchema: []string{
					base64.StdEncoding.EncodeToString([]byte("mydb")),
					base64.StdEncoding.EncodeToString([]byte("tidb_cdc")),
					base64.StdEncoding.EncodeToString([]byte("other")),
				},
				TableName: []string{
					base64.StdEncoding.EncodeToString([]byte("t1")),
					base64.StdEncoding.EncodeToString([]byte("t2")),
					base64.StdEncoding.EncodeToString([]byte("t3")),
				},
			},
			want:    []string{"mydb.t1", "other.t3"},
			wantErr: false,
		},
		{
			name: "invalid base64 schema",
			info: &TableSchemaInfo{
				TableSchema: []string{"!!!invalid-base64!!!"},
				TableName:   []string{base64.StdEncoding.EncodeToString([]byte("t1"))},
			},
			want:    nil,
			wantErr: true,
		},
		{
			name: "invalid base64 table name",
			info: &TableSchemaInfo{
				TableSchema: []string{base64.StdEncoding.EncodeToString([]byte("db"))},
				TableName:   []string{"!!!invalid-base64!!!"},
			},
			want:    nil,
			wantErr: true,
		},
		{
			name: "empty input",
			info: &TableSchemaInfo{
				TableSchema: []string{},
				TableName:   []string{},
			},
			want:    nil,
			wantErr: false,
		},
		{
			name: "schema/table length mismatch",
			info: &TableSchemaInfo{
				TableSchema: []string{base64.StdEncoding.EncodeToString([]byte("db"))},
				TableName:   []string{},
			},
			want:    nil,
			wantErr: true,
		},
	}

	for _, tt := range tests {
		t.Run(tt.name, func(t *testing.T) {
			got, err := parseQueryTable(tt.info, filter)
			if tt.wantErr {
				assert.Error(t, err)
			} else {
				assert.NoError(t, err)
				assert.Equal(t, tt.want, got)
			}
		})
	}
}

func TestFilterAgentTables_LengthMismatch(t *testing.T) {
	filter, err := tfilter.Parse([]string{"*.*"})
	require.NoError(t, err)

	tests := []struct {
		name    string
		result  *clusteragent.TablesEligibilityResult
		want    []string
		wantErr bool
	}{
		{
			name: "schemas longer than tables",
			result: &clusteragent.TablesEligibilityResult{
				Schemas: []string{"db1", "db2"},
				Tables:  []string{"t1"},
			},
			want:    nil,
			wantErr: true,
		},
		{
			name: "tables longer than schemas",
			result: &clusteragent.TablesEligibilityResult{
				Schemas: []string{"db1"},
				Tables:  []string{"t1", "t2"},
			},
			want:    nil,
			wantErr: true,
		},
		{
			name: "same length",
			result: &clusteragent.TablesEligibilityResult{
				Schemas: []string{"db1"},
				Tables:  []string{"t1"},
			},
			want:    []string{"db1.t1"},
			wantErr: false,
		},
		{
			name:    "nil result",
			result:  nil,
			want:    nil,
			wantErr: false,
		},
	}

	for _, tt := range tests {
		t.Run(tt.name, func(t *testing.T) {
			got, err := filterAgentTables(filter, tt.result)
			if tt.wantErr {
				assert.Error(t, err)
				return
			}

			require.NoError(t, err)
			assert.Equal(t, tt.want, got)
		})
	}
}