package impl

import (
	"context"
	"strconv"
	"time"

	"github.com/AlekSi/pointer"
	ticdc_v1 "github.com/tidbcloud/cluster-service-ng/api/tidbcloud/ticdc-service/v1"
	"github.com/tidbcloud/dataflow-service-ng/app/config"
	"github.com/tidbcloud/dataflow-service-ng/app/utils"
	dataflow "github.com/tidbcloud/dataflow-service-ng/idl/pbgen/proto/consoleapi"
	regionalapi "github.com/tidbcloud/dataflow-service-ng/pkg/regional-api"
	"github.com/tidbcloud/pkgv2/bizerr"
	"github.com/tidbcloud/pkgv2/clusterproxy"
)

const endpointLength = 2

var DownstreamDisplayNameMapping = map[dataflow.ChangefeedType]string{
	dataflow.ChangefeedType_KAFKA:           "Kafka",
	dataflow.ChangefeedType_MYSQL:           "MySQL",
	dataflow.ChangefeedType_SERVERLESS_TIDB: "ServerlessTiDB",
	dataflow.ChangefeedType_S3:              "S3",
	dataflow.ChangefeedType_GCS:             "GCS",
	dataflow.ChangefeedType_PULSAR:          "PULSAR",
}

func parseStartTSO(startPosition *dataflow.StartPosition) (uint64, bizerr.BizError) {
	if startPosition == nil {
		return 0, nil
	}

	switch startPosition.GetMode() {
	case dataflow.StartMode_FROM_NOW:
		return 0, nil
	case dataflow.StartMode_FROM_TSO:
		return startPosition.GetTso(), nil
	case dataflow.StartMode_FROM_UTC:
		startTime, err := time.Parse(time.RFC3339, startPosition.GetUtc())
		if err != nil {
			return 0, bizerr.New(bizerr.ErrCodeParams, "invalid start tso utc format, err: %s", err.Error())
		}
		return utils.GoTimeToTSO(startTime), nil
	default:
		return 0, bizerr.New(bizerr.ErrCodeParams, "invalid start position mode")
	}
}

func convertEventFilterRules(filterRules []*dataflow.EventFilterRule) []clusterproxy.EventFilterRule {
	dstRules := make([]clusterproxy.EventFilterRule, 0, len(filterRules))
	for _, rule := range filterRules {
		dstRules = append(dstRules, clusterproxy.EventFilterRule{
			Matcher:                  rule.GetMatcher(),
			IgnoreEvent:              rule.GetIgnoreEvent(),
			IgnoreSQL:                rule.GetIgnoreSql(),
			IgnoreInsertValueExpr:    rule.GetIgnoreInsertValueExpr(),
			IgnoreUpdateOldValueExpr: rule.GetIgnoreUpdateOldValueExpr(),
			IgnoreUpdateNewValueExpr: rule.GetIgnoreUpdateNewValueExpr(),
			IgnoreDeleteValueExpr:    rule.GetIgnoreDeleteValueExpr(),
		})
	}
	return dstRules
}

func checkStartTSO(
	ctx context.Context, preCheckCli regionalapi.TiCDCPreCheckClient, cluster *utils.Cluster, startTSO uint64,
) bizerr.BizError {
	region := &regionalapi.Region{SharedPoolID: cluster.GetSharedPoolId()}
	req := &ticdc_v1.ValidateChangefeedParametersRequest{
		ClusterId: strconv.FormatUint(cluster.GetClusterID(), 10),
		StartTso:  startTSO,
	}
	if err := preCheckCli.ValidateChangefeedParameters(ctx, region, req); err != nil {
		return bizerr.New(config.ErrCodeCDCInvalidParam, "failed to validate start tso")
	}
	return nil
}

func assembleChangefeedSinkConfigTableConfig(
	tableConfig *dataflow.TableConfig, replicaConfig *clusterproxy.ReplicaConfig,
) {
	forceReplicate := tableConfig.GetMode() == dataflow.TableMode_FORCE_SYNC
	ignoreIneligibleTable := tableConfig.GetMode() == dataflow.TableMode_IGNORE_NOT_SUPPORT_TABLE
	replicaConfig.Filter = &clusterproxy.FilterConfig{
		Rules:            tableConfig.GetFilterRule(),
		EventFilters:     convertEventFilterRules(tableConfig.GetEventFilterRule()),
		IgnoreTxnStartTs: tableConfig.GetIgnoreTxnStartTs(),
	}
	replicaConfig.ForceReplicate = pointer.ToBool(forceReplicate)
	replicaConfig.IgnoreIneligibleTable = pointer.ToBool(ignoreIneligibleTable)
	if tableConfig.BdrMode != nil {
		replicaConfig.BDRMode = pointer.ToBool(tableConfig.GetBdrMode())
	}
}