package impl

import (
	"time"

	"github.com/AlekSi/pointer"
	"github.com/tidbcloud/dataflow-service/app/utils"
	"github.com/tidbcloud/dataflow-service/idl/pbgen/proto/dataflow"
	"github.com/tidbcloud/pkgv2/bizerr"
	"github.com/tidbcloud/pkgv2/clusterproxy"
)

const endpointLength = 2

var DownstreamDisplayNameMapping = map[dataflow.ConnectorType]string{
	dataflow.ConnectorType_KAFKA:              "Kafka",
	dataflow.ConnectorType_MYSQL:              "MySQL",
	dataflow.ConnectorType_SERVERLESS_TIDB:    "ServerlessTiDB",
	dataflow.ConnectorType_S3:                 "S3",
	dataflow.ConnectorType_GCS:                "GCS",
	dataflow.ConnectorType_AZURE_BLOB_STORAGE: "AzureBlobStorage",
	dataflow.ConnectorType_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 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())
	}
}