package service

import (
	"context"
	"encoding/base64"
	"errors"
	"fmt"
	"net"
	"net/http"
	"sort"
	"strconv"

	filterutil "github.com/pingcap/tidb/pkg/util/filter"
	tfilter "github.com/pingcap/tidb/pkg/util/table-filter"
	apiv1 "github.com/tidbcloud/cluster-service-ng/api/tidbcloud/ticdc-service/v1"
	"github.com/tidbcloud/service-proxy/pkg/checker/core"
	"github.com/tidbcloud/tiinf/monitor/tilog"
	"go.uber.org/zap"
	"golang.org/x/sync/errgroup"
	"google.golang.org/grpc/codes"
	"google.golang.org/grpc/status"
	"google.golang.org/protobuf/types/known/emptypb"

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

	"github.com/tidbcloud/ticdc-service-ng/pkg/clusteragent"
	"github.com/tidbcloud/ticdc-service-ng/pkg/clusterproxy"
	"github.com/tidbcloud/ticdc-service-ng/pkg/clusterproxy/api"
)

const (
	// QueryEligibleSchemaTableClaus is the query clause to get tables that can be replicated by TiCDC
	QueryEligibleSchemaTableClaus = `
SELECT table_schema, table_name
FROM information_schema.tables
WHERE (table_schema, table_name) IN
      (SELECT table_schema, table_name
       FROM information_schema.columns
       WHERE COLUMN_KEY = 'PRI'
          OR COLUMN_KEY = 'UNI'
       union all
       SELECT table_schema, table_name
       FROM information_schema.tidb_indexes
       WHERE NON_UNIQUE = 0)
  AND table_schema NOT IN ('SYS', 'MYSQL', 'HEARTBEAT', 'DM_HEARTBEAT',
                           'INFORMATION_SCHEMA', 'PERFORMANCE_SCHEMA',
                           'METRICS_SCHEMA', 'INSPECTION_SCHEMA', 'sys', 'mysql', 'heartbeat',
                           'dm_heartbeat', 'information_schema', 'performance_schema',
                           'metrics_schema', 'inspection_schema');
`
	// QueryIneligibleSchemaTableClaus is the query clause to get tables that can not be replicated by TiCDC
	QueryIneligibleSchemaTableClaus = `
SELECT table_schema, table_name
FROM information_schema.tables
WHERE (table_schema, table_name) NOT IN
      (select distinct table_schema, table_name
       from (SELECT table_schema, table_name
             FROM information_schema.columns
             WHERE COLUMN_KEY = 'PRI'
                OR COLUMN_KEY = 'UNI'
             union all
             SELECT table_schema, table_name
             FROM information_schema.tidb_indexes
             where NON_UNIQUE = 0) d)
  AND table_schema NOT IN ('SYS', 'MYSQL', 'HEARTBEAT', 'DM_HEARTBEAT',
                           'INFORMATION_SCHEMA', 'PERFORMANCE_SCHEMA',
                           'METRICS_SCHEMA', 'INSPECTION_SCHEMA', 'sys', 'mysql', 'heartbeat',
                           'dm_heartbeat', 'information_schema', 'performance_schema',
                           'metrics_schema', 'inspection_schema');
`

	// TiCDCSystemSchema is the schema only use by TiCDC.
	TiCDCSystemSchema = "tidb_cdc"
	// LightningTaskInfoSchema is the schema only generated by Lightning
	LightningTaskInfoSchema = "lightning_task_info"

	// test connection types
	connectionTypeKafka  = "kafka"
	connectionTypeMysql  = "mysql"
	connectionTypeS3     = "s3"
	connectionTypeGCS    = "gcs"
	connectionTypePulsar = "pulsar"
)

// TableSchemaInfo is the response structure from the legacy clusterproxy SQL query path.
// Fields are base64-encoded by the proxy.
type TableSchemaInfo struct {
	TableSchema []string `json:"table_schema"`
	TableName   []string `json:"table_name"`
}

type tableVerifier interface {
	VerifyTable(
		ctx context.Context,
		logger *zap.Logger,
		req *apiv1.VerifyTableRequest,
		tblFilter tfilter.Filter,
	) (*apiv1.VerifyTableResponse, error)
}

type preCheckServerImpl struct {
	proxyModel    clusterproxy.Model
	tableVerifier tableVerifier
}

type preCheckServerOptions struct {
	proxyModel    clusterproxy.Model
	tableVerifier tableVerifier
}

// PreCheckServerOption customizes pre-check server construction.
type PreCheckServerOption func(*preCheckServerOptions) error

type proxyTableVerifier struct {
	proxyModel clusterproxy.Model
}

type agentTableVerifier struct {
	proxyModel        clusterproxy.Model
	clusterAgentModel clusteragent.Model
}

func defaultPreCheckServerOptions(proxyModel clusterproxy.Model) *preCheckServerOptions {
	return &preCheckServerOptions{
		proxyModel:    proxyModel,
		tableVerifier: &proxyTableVerifier{proxyModel: proxyModel},
	}
}

// WithClusterAgentModel switches VerifyTable to use the cluster-agent path.
func WithClusterAgentModel(agentModel clusteragent.Model) PreCheckServerOption {
	return func(opts *preCheckServerOptions) error {
		if opts == nil {
			return fmt.Errorf("pre-check server options are nil")
		}
		if agentModel == nil {
			return fmt.Errorf("agent model is required for cluster-agent precheck server")
		}
		opts.tableVerifier = &agentTableVerifier{
			proxyModel:        opts.proxyModel,
			clusterAgentModel: agentModel,
		}
		return nil
	}
}

// NewPreCheckServiceServer creates a pre-check server with legacy SQL path by default.
// Use WithClusterAgentModel to switch VerifyTable to cluster-agent path.
func NewPreCheckServiceServer(
	proxyModel clusterproxy.Model,
	opts ...PreCheckServerOption,
) (apiv1.TiCDCPrecheckServiceServer, error) {
	if proxyModel == nil {
		return nil, fmt.Errorf("proxy model is required for pre-check server")
	}

	options := defaultPreCheckServerOptions(proxyModel)
	for _, opt := range opts {
		if opt == nil {
			continue
		}
		if err := opt(options); err != nil {
			return nil, err
		}
	}
	if options.tableVerifier == nil {
		return nil, fmt.Errorf("table verifier is required for pre-check server")
	}
	return &preCheckServerImpl{
		proxyModel:    proxyModel,
		tableVerifier: options.tableVerifier,
	}, nil
}

func (p *preCheckServerImpl) TestNetworkConnection(ctx context.Context, req *apiv1.TestNetworkConnectionRequest) (*emptypb.Empty, error) {
	logger := tilog.WithContextL(ctx).With(zap.String("cluster", req.ClusterId), zap.String("type", req.Type))
	logger.Info("Starting network connection test")

	proxy, err := p.proxyModel.GetClusterProxy(ctx, req.ClusterId, clusterproxy.CDCHeavyOperationTimeout)
	if err != nil {
		logger.Error("Failed to get proxy", zap.Error(err))
		return nil, status.Errorf(codes.Internal, "Failed to get proxy for cluster %s: %s", req.ClusterId, err)
	}

	checkRequest, err := buildCheckRequest(req)
	if err != nil {
		logger.Error("Failed to build check request", zap.Error(err))
		return nil, status.Errorf(codes.InvalidArgument, "invalid request: %s", err)
	}

	if err := proxy.ChangefeedClient().TestConnection(ctx, checkRequest); err != nil {
		logger.Error("Failed to test connection", zap.Error(err))
		return nil, status.Errorf(codes.Internal, "connection test failed: %s", err)
	}

	logger.Info("Network connection test succeeded")
	return &emptypb.Empty{}, nil
}

func (p *preCheckServerImpl) ValidateChangefeedParameters(ctx context.Context, req *apiv1.ValidateChangefeedParametersRequest) (*emptypb.Empty, error) {
	logger := tilog.WithContextL(ctx).With(zap.String("cluster", req.ClusterId), zap.String("type", req.Type))
	logger.Info("Starting changefeed parameters validation")

	proxy, err := p.proxyModel.GetClusterProxy(ctx, req.ClusterId, clusterproxy.CDCHeavyOperationTimeout)
	if err != nil {
		logger.Error("Failed to get proxy", zap.Error(err))
		return nil, status.Errorf(codes.Internal, "Failed to get proxy for cluster %s: %s", req.ClusterId, err)
	}

	// if the start ts is 0, it means the changefeed is a new changefeed replicating data from the latest ts, do not need to check the safepoint
	if req.StartTso > 0 {
		safePoint, err := proxy.PDClient().GetPDSafepoint(ctx)
		if err != nil {
			logger.Error("Failed to get PD safepoint", zap.Error(err))
			return nil, status.Errorf(codes.Internal, "failed to get PD safepoint: %s", err)
		}
		if safePoint.SP > req.StartTso {
			logger.Error("StartTso is less than PD safepoint",
				zap.Uint64("startTso", req.StartTso),
				zap.Uint64("safepoint", safePoint.SP))
			return nil, status.Error(codes.InvalidArgument, "StartTso is less than PD safepoint")
		}
	}

	if req.KafkaParam == nil {
		logger.Info("Changefeed parameters validation succeeded (no kafka params)")
		return &emptypb.Empty{}, nil
	}

	checkRequest, err := buildKafkaCheckRequest(req)
	if err != nil {
		logger.Error("Failed to build kafka check request", zap.Error(err))
		return nil, status.Errorf(codes.InvalidArgument, "invalid kafka parameters: %s", err)
	}

	if err := proxy.ChangefeedClient().TestConnection(ctx, checkRequest); err != nil {
		logger.Error("Failed to validate changefeed parameter", zap.Error(err))
		return nil, status.Errorf(codes.Internal, "kafka connection validation failed: %s", err)
	}

	logger.Info("Changefeed parameters validation succeeded")
	return &emptypb.Empty{}, nil
}

func (p *preCheckServerImpl) VerifyTable(ctx context.Context, req *apiv1.VerifyTableRequest) (*apiv1.VerifyTableResponse, error) {
	logger := tilog.WithContextL(ctx).With(zap.String("cluster", req.TidbClusterId))
	filterRules := req.FilterRules
	if len(filterRules) == 0 {
		filterRules = []string{"*.*"}
	}

	logger.Info("verify table ", zap.Strings("filter", filterRules))
	tblFilter, err := tfilter.Parse(filterRules)
	if err != nil {
		return nil, status.Errorf(codes.InvalidArgument, "Failed to parse filters: %s", err)
	}
	if !req.CaseSensitive {
		tblFilter = tfilter.CaseInsensitive(tblFilter)
	}

	if p.tableVerifier == nil {
		return nil, status.Error(codes.Internal, "table verifier is not configured")
	}
	return p.tableVerifier.VerifyTable(ctx, logger, req, tblFilter)
}

type clusterAgentOpError struct {
	op  string
	err error
}

func (e *clusterAgentOpError) Error() string {
	if e == nil {
		return "<nil>"
	}
	if e.op == "" {
		return "cluster-agent request failed"
	}
	return fmt.Sprintf("cluster-agent %s failed", e.op)
}

func (e *clusterAgentOpError) Unwrap() error {
	if e == nil {
		return nil
	}
	return e.err
}

func (e *clusterAgentOpError) Cause() error {
	if e == nil {
		return nil
	}
	return e.err
}

func grpcCodeFromHTTPStatus(statusCode int) codes.Code {
	switch statusCode {
	case http.StatusBadRequest:
		return codes.InvalidArgument
	case http.StatusUnauthorized:
		return codes.Unauthenticated
	case http.StatusForbidden:
		return codes.PermissionDenied
	case http.StatusNotFound:
		return codes.NotFound
	case http.StatusRequestTimeout:
		return codes.DeadlineExceeded
	case http.StatusTooManyRequests:
		return codes.ResourceExhausted
	default:
		if statusCode >= 500 && statusCode <= 599 {
			return codes.Unavailable
		}
		return codes.Internal
	}
}

func mapClusterAgentError(err error) error {
	if err == nil {
		return nil
	}

	op := "request"
	var opErr *clusterAgentOpError
	if errors.As(err, &opErr) && opErr.op != "" {
		op = opErr.op
	}

	if errors.Is(err, context.Canceled) {
		return status.Errorf(codes.Canceled, "cluster-agent %s canceled", op)
	}
	if errors.Is(err, context.DeadlineExceeded) {
		return status.Errorf(codes.DeadlineExceeded, "cluster-agent %s timed out", op)
	}

	var httpErr *util.StatusError
	if errors.As(err, &httpErr) {
		return status.Errorf(grpcCodeFromHTTPStatus(httpErr.StatusCode),
			"cluster-agent %s failed: status=%d", op, httpErr.StatusCode)
	}

	var netErr net.Error
	if errors.As(err, &netErr) {
		if netErr.Timeout() {
			return status.Errorf(codes.DeadlineExceeded, "cluster-agent %s timed out", op)
		}
		return status.Errorf(codes.Unavailable, "cluster-agent %s failed: network error", op)
	}

	return status.Errorf(codes.Internal, "cluster-agent %s failed", op)
}

func logClusterAgentHTTPStatusError(logger *zap.Logger, err error) {
	if err == nil || logger == nil {
		return
	}

	var opErr *clusterAgentOpError
	if errors.As(err, &opErr) {
		err = opErr.err
	}

	var httpErr *util.StatusError
	if errors.As(err, &httpErr) {
		logger.Error("cluster-agent request failed",
			zap.Int("status", httpErr.StatusCode),
			zap.Error(err))
		logger.Debug("cluster-agent response body", zap.String("body", httpErr.TruncatedBody(1024)))
	}
}

// VerifyTable uses the new cluster-agent path:
// GetPublicKey → RSA encrypt credentials → QueryTiCDCTablesEligibility → PD safepoint
func (v *agentTableVerifier) VerifyTable(
	ctx context.Context,
	logger *zap.Logger,
	req *apiv1.VerifyTableRequest,
	tblFilter tfilter.Filter,
) (*apiv1.VerifyTableResponse, error) {
	if v.clusterAgentModel == nil {
		return nil, status.Errorf(codes.FailedPrecondition, "cluster agent model is not configured")
	}

	agentCli, err := v.clusterAgentModel.GetClient(ctx, req.TidbClusterId, clusterproxy.CDCHeavyOperationTimeout)
	if err != nil {
		logger.Error("Failed to get cluster agent client", zap.Error(err))
		if _, ok := status.FromError(err); ok {
			return nil, err
		}
		return nil, status.Errorf(codes.Internal, "Failed to get cluster agent client for cluster %s: %s", req.TidbClusterId, err)
	}

	pubKey, err := agentCli.GetPublicKey(ctx)
	if err != nil {
		opErr := &clusterAgentOpError{op: "get public key", err: err}
		logger.Error("Failed to get public key", zap.Error(err))
		logClusterAgentHTTPStatusError(logger, opErr)
		return nil, mapClusterAgentError(opErr)
	}

	encUser, err := clusteragent.EncryptWithPublicKey(pubKey, []byte(req.User))
	if err != nil {
		logger.Error("Failed to encrypt username", zap.Error(err))
		return nil, status.Errorf(codes.Internal, "Failed to encrypt credentials: %s", err)
	}
	encPwd, err := clusteragent.EncryptWithPublicKey(pubKey, []byte(req.Password))
	if err != nil {
		logger.Error("Failed to encrypt password", zap.Error(err))
		return nil, status.Errorf(codes.Internal, "Failed to encrypt credentials: %s", err)
	}

	g, gctx := errgroup.WithContext(ctx)
	var eligibleTables []string
	var ineligibleTables []string
	var gcSafePoint uint64

	g.Go(func() error {
		eligibleResult, ineligibleResult, err := agentCli.QueryTiCDCTablesEligibility(gctx, req.TidbClusterId, encUser, encPwd)
		if err != nil {
			return &clusterAgentOpError{op: "query tables eligibility", err: err}
		}
		eligible, err := filterAgentTables(tblFilter, eligibleResult)
		if err != nil {
			return &clusterAgentOpError{op: "filter eligible tables", err: err}
		}
		ineligible, err := filterAgentTables(tblFilter, ineligibleResult)
		if err != nil {
			return &clusterAgentOpError{op: "filter ineligible tables", err: err}
		}
		eligibleTables = eligible
		ineligibleTables = ineligible
		return nil
	})
	g.Go(func() error {
		proxy, err := v.proxyModel.GetClusterProxy(gctx, req.TidbClusterId, clusterproxy.CDCHeavyOperationTimeout)
		if err != nil {
			if errors.Is(err, context.Canceled) || errors.Is(err, context.DeadlineExceeded) {
				return nil
			}
			logger.Error("Failed to get proxy for PD safepoint", zap.Error(err))
			return nil
		}
		if proxy == nil {
			logger.Warn("GetClusterProxy returned nil, skipping PD safepoint retrieval")
			return nil
		}

		safepointResp, err := proxy.PDClient().GetPDSafepoint(gctx)
		if err != nil {
			if errors.Is(err, context.Canceled) || errors.Is(err, context.DeadlineExceeded) {
				return nil
			}
			logger.Error("Failed to get PD safepoint", zap.Error(err))
			return nil
		}
		if safepointResp != nil {
			gcSafePoint = safepointResp.SP
		}
		return nil
	})

	if err := g.Wait(); err != nil {
		var opErr *clusterAgentOpError
		if errors.As(err, &opErr) {
			logger.Error("Cluster-agent VerifyTable failed", zap.String("op", opErr.op), zap.Error(opErr.err))
		} else {
			logger.Error("Cluster-agent VerifyTable failed", zap.Error(err))
		}
		logClusterAgentHTTPStatusError(logger, err)
		return nil, mapClusterAgentError(err)
	}

	sort.Strings(eligibleTables)
	sort.Strings(ineligibleTables)

	logger.Info("verify table finished",
		zap.Int("ineligibleTables", len(ineligibleTables)),
		zap.Int("eligibleTables", len(eligibleTables)),
	)

	return &apiv1.VerifyTableResponse{
		IneligibleTables: ineligibleTables,
		EligibleTables:   eligibleTables,
		GcSafePoint:      gcSafePoint,
	}, nil
}

// VerifyTable uses the legacy clusterproxy path:
// GetClusterProxy → SQLClient().QuerySQL (base64 encoded results) → PD safepoint
func (v *proxyTableVerifier) VerifyTable(
	ctx context.Context,
	logger *zap.Logger,
	req *apiv1.VerifyTableRequest,
	tblFilter tfilter.Filter,
) (*apiv1.VerifyTableResponse, error) {
	proxy, err := v.proxyModel.GetClusterProxy(ctx, req.TidbClusterId, clusterproxy.CDCHeavyOperationTimeout)
	if err != nil {
		logger.Error("Failed to get proxy", zap.Error(err))
		return nil, status.Errorf(codes.Internal, "Failed to get proxy for cluster %s: %s", req.TidbClusterId, err)
	}

	eligibleTables, err := queryAndFilterTableSchema(ctx, tblFilter, proxy.SQLClient(), req.User, req.Password, QueryEligibleSchemaTableClaus)
	if err != nil {
		logger.Error("failed query eligible schema table", zap.Error(err))
		return nil, status.Errorf(codes.Internal, "Failed to query eligible schema table: %s", err)
	}

	ineligibleTable, err := queryAndFilterTableSchema(ctx, tblFilter, proxy.SQLClient(), req.User, req.Password, QueryIneligibleSchemaTableClaus)
	if err != nil {
		logger.Error("failed query ineligible schema table", zap.Error(err))
		return nil, status.Errorf(codes.Internal, "failed query ineligible schema table: %s", err)
	}
	sort.Strings(eligibleTables)
	sort.Strings(ineligibleTable)
	logger.Info("verify table finished",
		zap.Int("ineligibleTable", len(ineligibleTable)),
		zap.Int("eligibleTables", len(eligibleTables)),
	)

	var gcSafePoint uint64
	safepointResp, err := proxy.PDClient().GetPDSafepoint(ctx)
	if err != nil {
		logger.Error("Failed to get PD safepoint", zap.Error(err))
	}
	if safepointResp != nil {
		gcSafePoint = safepointResp.SP
	}

	return &apiv1.VerifyTableResponse{
		IneligibleTables: ineligibleTable,
		EligibleTables:   eligibleTables,
		GcSafePoint:      gcSafePoint,
	}, nil
}

// queryAndFilterTableSchema queries table schemas via the legacy clusterproxy SQL path.
// Results are base64-encoded by the proxy and decoded locally.
func queryAndFilterTableSchema(
	ctx context.Context,
	filter tfilter.Filter,
	client api.SQLClient,
	user, password, sql string,
) ([]string, error) {
	var queryInfo TableSchemaInfo
	if err := client.QuerySQL(ctx, &api.ClusterOperator{
		Username:   user,
		AuthMethod: api.AuthMethodPassword,
		Credential: password,
	}, sql, &queryInfo); err != nil {
		return nil, err
	}
	return parseQueryTable(&queryInfo, filter)
}

func parseQueryTable(queryInfo *TableSchemaInfo, filter tfilter.Filter) ([]string, error) {
	if queryInfo == nil {
		return nil, fmt.Errorf("queryInfo is nil")
	}
	if len(queryInfo.TableSchema) != len(queryInfo.TableName) {
		return nil, fmt.Errorf("table schema/name length mismatch: schemas=%d names=%d", len(queryInfo.TableSchema), len(queryInfo.TableName))
	}
	if len(queryInfo.TableSchema) == 0 {
		return nil, nil
	}

	queryTable := make([]string, 0, len(queryInfo.TableSchema))
	for i := 0; i < len(queryInfo.TableSchema); i++ {
		schema := queryInfo.TableSchema[i]
		table := queryInfo.TableName[i]
		schemaRet, err := base64.StdEncoding.DecodeString(schema)
		if err != nil {
			return nil, err
		}
		tableRet, err := base64.StdEncoding.DecodeString(table)
		if err != nil {
			return nil, err
		}
		if !isSysSchema(string(schemaRet)) && filter.MatchTable(string(schemaRet), string(tableRet)) {
			queryTable = append(queryTable, fmt.Sprintf("%s.%s", schemaRet, tableRet))
		}
	}
	return queryTable, nil
}

// filterAgentTables applies local filtering for TiCDC-specific system schemas
// to schema/table pairs returned by the cluster-agent API.
func filterAgentTables(
	filter tfilter.Filter,
	result *clusteragent.TablesEligibilityResult,
) ([]string, error) {
	if result == nil {
		return nil, nil
	}
	schemas := result.Schemas
	tables := result.Tables
	if len(schemas) != len(tables) {
		return nil, fmt.Errorf("schema/table length mismatch: schemas=%d tables=%d", len(schemas), len(tables))
	}
	if len(schemas) == 0 {
		return nil, nil
	}

	filtered := make([]string, 0, len(schemas))
	for i := 0; i < len(schemas); i++ {
		schema := schemas[i]
		table := tables[i]
		if !isSysSchema(schema) && filter.MatchTable(schema, table) {
			filtered = append(filtered, fmt.Sprintf("%s.%s", schema, table))
		}
	}
	return filtered, nil
}

// buildCheckRequest builds a CheckRequest from TestNetworkConnectionRequest.
// Returns error if the request is invalid or missing required parameters.
func buildCheckRequest(req *apiv1.TestNetworkConnectionRequest) (*core.CheckRequest, error) {
	if req == nil {
		return nil, errors.New("request cannot be nil")
	}

	checkRequest := &core.CheckRequest{TargetType: req.Type}
	switch req.Type {
	case connectionTypeMysql:
		if req.MysqlParam == nil {
			return nil, errors.New("mysql parameters are required for mysql connection type")
		}
		host, port, err := parseHostPort(req.MysqlParam.Url)
		if err != nil {
			return nil, fmt.Errorf("invalid mysql url: %w", err)
		}
		checkRequest.MySQL = &core.MySQLParams{
			Host:     host,
			Port:     port,
			User:     req.MysqlParam.User,
			Password: req.MysqlParam.Password,
		}
	case connectionTypeKafka:
		if req.KafkaParam == nil {
			return nil, errors.New("kafka parameters are required for kafka connection type")
		}
		kafkaAuthType := core.KafkaAuthType(req.KafkaParam.AuthType)
		if req.KafkaParam.AuthType != "" && !kafkaAuthType.IsValid() {
			return nil, fmt.Errorf("invalid kafka authentication type: %q", req.KafkaParam.AuthType)
		}
		checkRequest.Kafka = &core.KafkaParams{
			Brokers:   req.KafkaParam.BoostrapEndpoints,
			User:      req.KafkaParam.User,
			Password:  req.KafkaParam.Password,
			EnableTLS: req.KafkaParam.EnableTls,
			AuthType:  req.KafkaParam.AuthType,
		}
	case connectionTypeS3:
		if req.S3Param == nil {
			return nil, errors.New("s3 parameters are required for s3 connection type")
		}
		checkRequest.S3 = &core.S3Params{
			URI:      req.S3Param.S3Uri,
			Endpoint: req.S3Param.Endpoint,
			AWSAuth: core.AWSAuth{
				AccessKeyID:     req.S3Param.AccessKeyId,
				SecretAccessKey: req.S3Param.SecretAccessKey,
				RoleARN:         req.S3Param.RoleArn,
				ExternalID:      req.S3Param.ExternalId,
			},
		}
	case connectionTypeGCS:
		if req.GcsParam == nil {
			return nil, errors.New("gcs parameters are required for gcs connection type")
		}
		checkRequest.GCS = &core.GCSParams{
			Url:                       req.GcsParam.GcsUrl,
			Region:                    req.GcsParam.Region,
			ImpersonateServiceAccount: req.GcsParam.ImpersonateServiceAccount,
		}
	case connectionTypePulsar:
		if req.GetPulsarParam() == nil {
			return nil, errors.New("pulsar parameters are required for pulsar connection type")
		}
		pulsarAuthType := core.PulsarAuthType(req.GetPulsarParam().GetAuthenticationType())
		if !pulsarAuthType.IsValid() {
			return nil, fmt.Errorf("invalid pulsar authentication type: %q, supported: basic, token, or empty",
				req.GetPulsarParam().GetAuthenticationType())
		}
		checkRequest.Pulsar = &core.PulsarParams{
			ServiceURL:        req.GetPulsarParam().GetServiceUrl(),
			AuthType:          pulsarAuthType,
			Username:          req.GetPulsarParam().GetUsername(),
			Password:          req.GetPulsarParam().GetPassword(),
			Token:             req.GetPulsarParam().GetToken(),
			ConnectionTimeout: req.GetPulsarParam().GetConnectionTimeout(),
			Tenant:            req.GetPulsarParam().GetTenant(),
			Namespace:         req.GetPulsarParam().GetNamespace(),
			Topic:             req.GetPulsarParam().GetTopic(),
		}
	default:
		return nil, fmt.Errorf("unsupported connection type: %s", req.Type)
	}
	return checkRequest, nil
}

// buildKafkaCheckRequest builds a CheckRequest for Kafka from ValidateChangefeedParametersRequest.
// Returns error if the kafka parameters are missing.
func buildKafkaCheckRequest(req *apiv1.ValidateChangefeedParametersRequest) (*core.CheckRequest, error) {
	if req == nil {
		return nil, errors.New("request cannot be nil")
	}
	if req.KafkaParam == nil {
		return nil, errors.New("kafka parameters are required")
	}

	kafkaAuthType := core.KafkaAuthType(req.KafkaParam.AuthType)
	if req.KafkaParam.AuthType != "" && !kafkaAuthType.IsValid() {
		return nil, fmt.Errorf("invalid kafka authentication type: %q", req.KafkaParam.AuthType)
	}

	return &core.CheckRequest{
		TargetType: req.Type,
		Kafka: &core.KafkaParams{
			Brokers:                 req.KafkaParam.Url,
			User:                    req.KafkaParam.User,
			Password:                req.KafkaParam.Password,
			EnableTLS:               req.KafkaParam.EnableTls,
			AuthType:                req.KafkaParam.AuthType,
			ReplicationFactor:       req.KafkaParam.ReplicationFactor,
			PartitionNum:            req.KafkaParam.PartitionNum,
			SchemaRegistryEndpoints: req.KafkaParam.SchemaRegistryEndpoints,
		},
	}, nil
}

// parseHostPort parses a host:port string to extract host and port.
// Supports IPv4, IPv6 (with brackets), and hostname formats.
// Returns default MySQL port 3306 if port is not specified.
// Returns an error if an invalid port is explicitly provided.
func parseHostPort(addr string) (string, int, error) {
	const defaultPort = 3306

	if addr == "" {
		return "", defaultPort, nil
	}

	// Try net.SplitHostPort first (handles IPv6 with brackets)
	host, portStr, err := net.SplitHostPort(addr)
	if err != nil {
		// No port specified, return addr as host with default port
		return addr, defaultPort, nil
	}

	port, err := strconv.Atoi(portStr)
	if err != nil {
		return host, 0, fmt.Errorf("invalid port %q: not a number", portStr)
	}
	if port < 1 || port > 65535 {
		return host, 0, fmt.Errorf("invalid port %d: must be between 1 and 65535", port)
	}

	return host, port, nil
}

// isSysSchema returns true if the given schema is a system schema
func isSysSchema(db string) bool {
	switch db {
	// TiCDCSystemSchema is used by TiCDC only.
	// Tables in TiCDCSystemSchema should not be replicated by cdc.
	case TiCDCSystemSchema:
		return true
	// Tables in LightningTaskInfoSchema should not be replicated by cdc.
	case LightningTaskInfoSchema:
		return true
	default:
		return filterutil.IsSystemSchema(db)
	}
}