package domain

import (
	"fmt"
	"regexp"
	"strconv"
	"strings"

	"github.com/tidbcloud/dataflow-service/app/repo/entity"
	"github.com/tidbcloud/dataflow-service/internal/config"
	encryptor_wrapper "github.com/tidbcloud/dataflow-service/pkg/encryptor-wrapper"
	replay_v1 "github.com/tidbcloud/infra-provider/api/tidbcloud/cluster-server/v1"
	"github.com/tidbcloud/pkgv2/bizerr"
	mgmt_commonv1 "github.com/tidbcloud/tidb-management-service/api/spec/global/common/v1"
	mgmt "github.com/tidbcloud/tidb-management-service/api/spec/tidb_mgmt_service/v1"
)

func ToReplayProto(
	replay *entity.Replay,
	sourceCluster *mgmt.Cluster,
	targetCluster *mgmt.Cluster,
	state entity.ReplayDesiredState,
	encryptor encryptor_wrapper.Encryptor,
	deploymentProvider string,
	deploymentRegion string,
	targetEndpoint string,
) (*replay_v1.Replay, bizerr.BizError) {
	replayProto := &replay_v1.Replay{
		ReplayId:  BuildRegionalReplayID(replay.ID),
		TenantId:  strconv.FormatUint(replay.TenantID.Uint64(), 10),
		ProjectId: strconv.FormatUint(replay.ProjectID.Uint64(), 10),
		Provider:  deploymentProvider,
		Region:    deploymentRegion,
		State:     toReplayStateProto(state),
	}
	if replayProto.State == replay_v1.Replay_STATE_UNSPECIFIED {
		return nil, bizerr.New(bizerr.ErrCodeParams,
			"unsupported replay state, state: %s", state)
	}
	if targetEndpoint != "" {
		replayProto.PremiumTargetEndpoint = &targetEndpoint
	}

	if sourceCluster != nil {
		clusterType := toReplayClusterType(sourceCluster)
		if clusterType == replay_v1.Replay_CLUSTER_TYPE_UNSPECIFIED {
			return nil, bizerr.New(bizerr.ErrCodeParams,
				"unsupported source cluster type, cluster id: %d", sourceCluster.ClusterId)
		}
		replayProto.SourceCluster = &replay_v1.Replay_ClusterRef{
			ClusterType: clusterType,
			ClusterId:   strconv.FormatUint(sourceCluster.ClusterId, 10),
			Provider:    CloudProviderToInfraProvider(sourceCluster.CloudProvider),
			Region:      sourceCluster.Region,
			TenantId:    strconv.FormatUint(sourceCluster.OrganizationId, 10),
			ProjectId:   strconv.FormatUint(sourceCluster.ProjectId, 10),
		}
	}
	clusterType := toReplayClusterType(targetCluster)
	if clusterType == replay_v1.Replay_CLUSTER_TYPE_UNSPECIFIED {
		return nil, bizerr.New(bizerr.ErrCodeParams,
			"unsupported target cluster type, cluster id: %d", targetCluster.GetClusterId())
	}
	replayProto.TargetCluster = &replay_v1.Replay_ClusterRef{
		ClusterType: clusterType,
		ClusterId:   strconv.FormatUint(targetCluster.GetClusterId(), 10),
		Provider:    CloudProviderToInfraProvider(targetCluster.GetCloudProvider()),
		Region:      targetCluster.GetRegion(),
		TenantId:    strconv.FormatUint(targetCluster.GetOrganizationId(), 10),
		ProjectId:   strconv.FormatUint(targetCluster.GetProjectId(), 10),
	}

	toReplayDefaults(replayProto, replay, encryptor)
	toReplayTasks(replayProto, replay, encryptor)
	return replayProto, nil
}

// FromReplayStatusProto convert the replay_v1.ReplayStatus to the entity.ReplayStatus
func FromReplayStatusProto(protoStatus *replay_v1.ReplayStatus) *entity.ReplayStatus {
	if protoStatus == nil {
		return nil
	}

	status := &entity.ReplayStatus{
		Phase:   fromReplayPhaseProto(protoStatus.Phase),
		Message: protoStatus.Message,
		Tasks:   fromReplayTaskStatusListProto(protoStatus.Tasks),
	}

	if protoStatus.LastUpdateTime != nil {
		lastUpdateTime := protoStatus.LastUpdateTime.AsTime()
		status.LastUpdateTime = &lastUpdateTime
	}

	if protoStatus.AwsNodeRoleArn != "" {
		status.AWSNodeRoleArn = protoStatus.AwsNodeRoleArn
	}

	return status
}

func toReplayStateProto(state entity.ReplayDesiredState) replay_v1.Replay_State {
	switch state {
	case entity.ReplayDesiredStateDeploy:
		return replay_v1.Replay_STATE_DEPLOY
	case entity.ReplayDesiredStateActive:
		return replay_v1.Replay_STATE_ACTIVE
	case entity.ReplayDesiredStatePause:
		return replay_v1.Replay_STATE_PAUSE
	}
	return replay_v1.Replay_STATE_UNSPECIFIED
}

// toReplayDefaults fill the default configuration of the Replay
func toReplayDefaults(replayProto *replay_v1.Replay, replay *entity.Replay, encryptor encryptor_wrapper.Encryptor) {
	replayProto.Defaults = &replay_v1.Replay_ReplayDefaults{}

	if replay.Replicas > 0 {
		replayProto.Defaults.Replicas = &replay.Replicas
	}
	if replay.TaskResources != nil {
		replayProto.Defaults.Resources = &replay_v1.Replay_Resources{
			InstanceType: &replay.TaskResources.InstanceType,
			StorageSize:  &replay.TaskResources.StorageSize,
		}
	}
	if replay.ReplayImage != "" {
		replayProto.Defaults.Image = &replay.ReplayImage
	} else if config.Global().ReplayConfig.DefaultReplayImage != "" {
		replayProto.Defaults.Image = &config.Global().ReplayConfig.DefaultReplayImage
	}
	if replay.ReplayConfig != nil {
		replayProto.Defaults.Options = &replay_v1.Replay_ReplayOptions{
			ReplayStartTime: replay.ReplayConfig.ReplayConfig.ReplayStartTime,
			RunStartTime:    replay.ReplayConfig.ReplayConfig.RunStartTime,
			ExtraArgs:       replay.ReplayConfig.ReplayConfig.CustomArguments,
		}
		if replay.ReplayConfig.Username != "" {
			decryptPassword, _ := encryptor.DecryptBase64(replay.ReplayConfig.Password)
			replayProto.Defaults.Credential = &replay_v1.Replay_DatabaseCredential{
				Username: replay.ReplayConfig.Username,
				Password: decryptPassword,
			}
		}
	}
	if replay.SourceAuditCredential != nil {
		if replay.SourceAuditCredential.AWS != nil {
			replayProto.Defaults.SourceAuditLog = &replay_v1.Replay_AuditLogAddr{
				Source: &replay_v1.Replay_AuditLogAddr_Aws{
					Aws: &replay_v1.Replay_AWSAuditLog{
						S3Url:      replay.SourceAuditCredential.AWS.S3URL,
						RoleArn:    replay.SourceAuditCredential.AWS.RoleARN,
						ExternalId: replay.SourceAuditCredential.AWS.ExternalID,
					},
				},
			}
		}
		if replay.SourceAuditCredential.Aliyun != nil {
			replayProto.Defaults.SourceAuditLog = &replay_v1.Replay_AuditLogAddr{
				Source: &replay_v1.Replay_AuditLogAddr_Aliyun{
					Aliyun: &replay_v1.Replay_AliyunAuditLog{
						BucketUrl: replay.SourceAuditCredential.Aliyun.BucketURL,
					},
				},
			}
		}
	}
}

// toReplayTasks fill the task list of the Replay
func toReplayTasks(replayProto *replay_v1.Replay, replay *entity.Replay, encryptor encryptor_wrapper.Encryptor) {
	for _, task := range replay.ReplayTasks {
		replayProto.Tasks = append(replayProto.Tasks, toReplayTaskProto(&task, encryptor))
	}
}

// toReplayTaskProto convert the entity.ReplayTask to the replay_v1.Replay_ReplayTask
func toReplayTaskProto(task *entity.ReplayTask, encryptor encryptor_wrapper.Encryptor) *replay_v1.Replay_ReplayTask {
	taskProto := &replay_v1.Replay_ReplayTask{
		Name:   task.Name,
		Source: toReplayDataSourceProto(&task.DataSource),
		Target: toReplayDataTargetProto(&task.DataTarget, encryptor),
	}
	if task.Replicas > 0 {
		taskProto.Replicas = &task.Replicas
	}
	if task.Resources != nil {
		taskProto.Resources = &replay_v1.Replay_Resources{
			InstanceType: &task.Resources.InstanceType,
			StorageSize:  &task.Resources.StorageSize,
		}
	}
	if task.Image != "" {
		taskProto.Image = &task.Image
	}
	if task.ReplayStartTime != "" {
		taskProto.ReplayStartTime = &task.ReplayStartTime
	}
	if task.RunStartTime != "" {
		taskProto.RunStartTime = &task.RunStartTime
	}
	for _, perReplayConfig := range task.PerReplayConfigs {
		if perReplayConfig.ReplayIndex == nil {
			continue
		}
		replayReplicaConfig := &replay_v1.Replay_ReplicaConfig{
			ReplicaIndex: *perReplayConfig.ReplayIndex,
			Options: &replay_v1.Replay_ReplayOptions{
				ExtraArgs: perReplayConfig.CustomArguments,
			},
		}
		if perReplayConfig.ReplayStartTime != nil {
			replayReplicaConfig.Options.ReplayStartTime = perReplayConfig.ReplayStartTime
		}
		if perReplayConfig.RunStartTime != nil {
			replayReplicaConfig.Options.RunStartTime = perReplayConfig.RunStartTime
		}
		taskProto.ReplicaConfigs = append(taskProto.ReplicaConfigs, replayReplicaConfig)
	}
	return taskProto
}

// toReplayDataSourceProto convert the entity.ReplayDataSource to the replay_v1.Replay_DataSource
func toReplayDataSourceProto(dataSource *entity.ReplayDataSource) *replay_v1.Replay_DataSource {
	sourceAuditLogAddrs := make([]*replay_v1.Replay_AuditLogAddr, 0, len(dataSource.SourceAuditLogAddrs))
	for _, auditLogAddr := range dataSource.SourceAuditLogAddrs {
		if auditLogAddr.AWS != nil {
			sourceAuditLogAddrs = append(sourceAuditLogAddrs, &replay_v1.Replay_AuditLogAddr{
				Source: &replay_v1.Replay_AuditLogAddr_Aws{
					Aws: &replay_v1.Replay_AWSAuditLog{
						S3Url:      auditLogAddr.AWS.S3URL,
						RoleArn:    auditLogAddr.AWS.RoleARN,
						ExternalId: auditLogAddr.AWS.ExternalID,
					},
				},
			})
		}
		if auditLogAddr.Aliyun != nil {
			sourceAuditLogAddrs = append(sourceAuditLogAddrs, &replay_v1.Replay_AuditLogAddr{
				Source: &replay_v1.Replay_AuditLogAddr_Aliyun{
					Aliyun: &replay_v1.Replay_AliyunAuditLog{
						BucketUrl: auditLogAddr.Aliyun.BucketURL,
					},
				},
			})
		}
	}
	return &replay_v1.Replay_DataSource{
		AuditLogs: sourceAuditLogAddrs,
	}
}

// toReplayDataTargetProto convert the entity.ReplayDataTarget to the replay_v1.Replay_DataTarget
func toReplayDataTargetProto(
	dataTarget *entity.ReplayDataTarget, encryptor encryptor_wrapper.Encryptor,
) *replay_v1.Replay_DataTarget {
	dataTargetProto := &replay_v1.Replay_DataTarget{
		Endpoint: dataTarget.Endpoint,
	}
	if dataTarget.Username != nil && dataTarget.Password != nil {
		decryptPassword, _ := encryptor.DecryptBase64(*dataTarget.Password)
		dataTargetProto.Credential = &replay_v1.Replay_DatabaseCredential{
			Username: *dataTarget.Username,
			Password: decryptPassword,
		}
	}
	return dataTargetProto
}

// fromReplayTaskStatusListProto convert the replay_v1.ReplayStatus_ReplayTaskStatus list to the entity.ReplayTaskStatus list
func fromReplayTaskStatusListProto(protoTasks []*replay_v1.ReplayStatus_ReplayTaskStatus) []entity.ReplayTaskStatus {
	if len(protoTasks) == 0 {
		return nil
	}

	tasks := make([]entity.ReplayTaskStatus, 0, len(protoTasks))
	for _, protoTask := range protoTasks {
		tasks = append(tasks, fromReplayTaskStatusProto(protoTask))
	}
	return tasks
}

// fromReplayTaskStatusProto convert the replay_v1.ReplayStatus_ReplayTaskStatus to the entity.ReplayTaskStatus
func fromReplayTaskStatusProto(protoTask *replay_v1.ReplayStatus_ReplayTaskStatus) entity.ReplayTaskStatus {
	task := entity.ReplayTaskStatus{
		Name:            protoTask.Name,
		Message:         protoTask.Message,
		ReplicaStatuses: fromReplayReplicaStatusListProto(protoTask.ReplicaStatuses),
	}
	return task
}

// fromReplayReplicaStatusListProto convert the replay_v1.ReplayStatus_ReplayReplicaStatus list to the entity.ReplayReplicaStatus list
func fromReplayReplicaStatusListProto(
	protoReplicas []*replay_v1.ReplayStatus_ReplayReplicaStatus,
) []entity.ReplayReplicaStatus {
	if len(protoReplicas) == 0 {
		return nil
	}

	replicas := make([]entity.ReplayReplicaStatus, 0, len(protoReplicas))
	for _, protoReplica := range protoReplicas {
		replicas = append(replicas, fromReplayReplicaStatusProto(protoReplica))
	}
	return replicas
}

// fromReplayReplicaStatusProto convert the replay_v1.ReplayStatus_ReplayReplicaStatus to the entity.ReplayReplicaStatus
func fromReplayReplicaStatusProto(protoReplica *replay_v1.ReplayStatus_ReplayReplicaStatus) entity.ReplayReplicaStatus {
	replica := entity.ReplayReplicaStatus{
		PodName: protoReplica.PodName,
		Index:   protoReplica.Index,
		Phase:   fromReplayTaskPhaseProto(protoReplica.Phase),
		Ready:   protoReplica.Ready,
		Message: protoReplica.Message,
	}

	if protoReplica.TaskInfo != nil {
		replica.TaskInfo = fromReplayTaskInfoProto(protoReplica.TaskInfo)
	}

	return replica
}

// fromReplayTaskInfoProto convert the replay_v1.ReplayStatus_ReplayTaskInfo to the entity.ReplayTaskInfo
func fromReplayTaskInfoProto(protoInfo *replay_v1.ReplayStatus_ReplayTaskInfo) *entity.ReplayTaskInfo {
	if protoInfo == nil {
		return nil
	}

	info := &entity.ReplayTaskInfo{
		Type:                 protoInfo.Type,
		Status:               protoInfo.Status,
		StartTime:            protoInfo.StartTime,
		EndTime:              protoInfo.EndTime,
		Progress:             protoInfo.Progress,
		Error:                protoInfo.Error,
		LastCommandTimestamp: protoInfo.LastCommandTimestamp,
		SyncTime:             protoInfo.SyncTime,
		TotalCommands:        protoInfo.TotalCommands,
		ReplayedCommands:     protoInfo.ReplayedCommands,
		FailedCommands:       protoInfo.FailedCommands,
		TotalBytes:           protoInfo.TotalBytes,
		ProcessedBytes:       protoInfo.ProcessedBytes,
	}

	if len(protoInfo.SourceAuditLogAddrs) > 0 {
		info.SourceAuditLogAddrs = fromReplayAuditLogAddrListProto(protoInfo.SourceAuditLogAddrs)
	}

	return info
}

// fromReplayAuditLogAddrListProto convert the replay_v1.Replay_AuditLogAddr list to the entity.AuditLogAddr list
func fromReplayAuditLogAddrListProto(protoAddrs []*replay_v1.Replay_AuditLogAddr) []entity.AuditLogAddr {
	if len(protoAddrs) == 0 {
		return nil
	}

	addrs := make([]entity.AuditLogAddr, 0, len(protoAddrs))
	for _, protoAddr := range protoAddrs {
		if addr := fromReplayAuditLogAddrProto(protoAddr); addr != nil {
			addrs = append(addrs, *addr)
		}
	}
	return addrs
}

// fromReplayAuditLogAddrProto convert the replay_v1.Replay_AuditLogAddr to the entity.AuditLogAddr
func fromReplayAuditLogAddrProto(protoAddr *replay_v1.Replay_AuditLogAddr) *entity.AuditLogAddr {
	if protoAddr == nil {
		return nil
	}

	addr := &entity.AuditLogAddr{}

	if aws := protoAddr.GetAws(); aws != nil {
		addr.AWS = &entity.AWSAuditLog{
			S3URL:   aws.S3Url,
			RoleARN: aws.RoleArn,
		}
		if externalID := aws.GetExternalId(); externalID != "" {
			addr.AWS.ExternalID = &externalID
		}
	}
	if aliyun := protoAddr.GetAliyun(); aliyun != nil {
		addr.Aliyun = &entity.AliyunAuditLog{
			BucketURL: RedactAliyunBucketURLForTransport(aliyun.BucketUrl),
		}
	}

	return addr
}

// toReplayClusterType get the replay_v1.Replay_ClusterType from the mgmt.Cluster
func toReplayClusterType(clusterInfo *mgmt.Cluster) replay_v1.Replay_ClusterType {
	if clusterInfo == nil {
		return replay_v1.Replay_CLUSTER_TYPE_UNSPECIFIED
	}
	switch clusterInfo.DeployType {
	case mgmt.DeployType_DEPLOY_TYPE_DEDICATED:
		return replay_v1.Replay_CLUSTER_TYPE_DEDICATED
	case mgmt.DeployType_DEPLOY_TYPE_NEXTGEN:
		switch clusterInfo.GetNextGenCluster().GetServicePlan() {
		case mgmt_commonv1.ServicePlan_Premium:
			return replay_v1.Replay_CLUSTER_TYPE_PREMIUM
		case mgmt_commonv1.ServicePlan_Essential:
			return replay_v1.Replay_CLUSTER_TYPE_ESSENTIAL
		case mgmt_commonv1.ServicePlan_Starter:
			return replay_v1.Replay_CLUSTER_TYPE_STARTER
		}
	}
	return replay_v1.Replay_CLUSTER_TYPE_UNSPECIFIED
}

// fromReplayPhaseProto convert the replay_v1.ReplayStatus_Phase to the entity.ReplayPhase
func fromReplayPhaseProto(phase replay_v1.ReplayStatus_Phase) entity.ReplayPhase {
	switch phase {
	case replay_v1.ReplayStatus_PHASE_PENDING:
		return entity.ReplayPhasePending
	case replay_v1.ReplayStatus_PHASE_CREATING:
		return entity.ReplayPhaseCreating
	case replay_v1.ReplayStatus_PHASE_RUNNING:
		return entity.ReplayPhaseRunning
	case replay_v1.ReplayStatus_PHASE_STOPPED:
		return entity.ReplayPhaseStopped
	case replay_v1.ReplayStatus_PHASE_DELETING:
		return entity.ReplayPhaseDeleting
	default:
		return entity.ReplayPhaseUnspecified
	}
}

// fromReplayTaskPhaseProto convert the replay_v1.ReplayStatus_TaskPhase to the entity.ReplayTaskPhase
func fromReplayTaskPhaseProto(phase replay_v1.ReplayStatus_TaskPhase) entity.ReplayTaskPhase {
	switch phase {
	case replay_v1.ReplayStatus_TASK_PHASE_PENDING:
		return entity.ReplayTaskPhasePending
	case replay_v1.ReplayStatus_TASK_PHASE_RUNNING:
		return entity.ReplayTaskPhaseRunning
	case replay_v1.ReplayStatus_TASK_PHASE_STOPPED:
		return entity.ReplayTaskPhaseStopped
	case replay_v1.ReplayStatus_TASK_PHASE_DELETED:
		return entity.ReplayTaskPhaseDeleted
	default:
		return entity.ReplayTaskPhaseUnspecified
	}
}

func FromReplayStateProto(state replay_v1.Replay_State) entity.ReplayDesiredState {
	switch state {
	case replay_v1.Replay_STATE_DEPLOY:
		return entity.ReplayDesiredStateDeploy
	case replay_v1.Replay_STATE_ACTIVE:
		return entity.ReplayDesiredStateActive
	case replay_v1.Replay_STATE_PAUSE:
		return entity.ReplayDesiredStatePause
	}
	return ""
}

// CloudProviderToInfraProvider converts Dataflow-facing cloud provider names to
// infra-provider provider names.
func CloudProviderToInfraProvider(cloudProvider string) string {
	switch strings.ToLower(strings.TrimSpace(cloudProvider)) {
	case "aws":
		return string(InfraProviderTypeEKS)
	case "gcp":
		return string(InfraProviderTypeGKE)
	case "azure":
		return string(InfraProviderTypeAKS)
	case "aliyun", "alicloud", "ack":
		return string(InfraProviderTypeACK)
	}
	return ""
}

func BuildRegionalReplayID(replayID entity.ID) string {
	return fmt.Sprintf("replay-%d", replayID.Uint64())
}

var aliyunBucketURLCredentialPattern = regexp.MustCompile(
	`(?i)[?&](?:[^=&#]*access[-_]?key[^=&#]*|[^=&#]*secret[^=&#]*|[^=&#]*signature[^=&#]*|[^=&#]*credential[^=&#]*|[^=&#]*token[^=&#]*|sig)=`,
)

// RedactAliyunBucketURLForTransport strips query and fragment only when the URL
// contains credential-like query parameters, preserving safe locator queries.
// Examples:
//   - s3://bucket/path?access-key=ak&secret-access-key=sk#frag -> s3://bucket/path
//   - s3://bucket/path?endpoint=oss -> unchanged
func RedactAliyunBucketURLForTransport(bucketURL string) string {
	if !aliyunBucketURLCredentialPattern.MatchString(bucketURL) {
		return bucketURL
	}
	if idx := strings.IndexAny(bucketURL, "?#"); idx >= 0 {
		return bucketURL[:idx]
	}
	return bucketURL
}