package cvt

import (
	"maps"

	"google.golang.org/protobuf/types/known/timestamppb"
	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"

	cloudv1 "github.com/tidbcloud/infra-api/api/v1"
	apiv1 "github.com/tidbcloud/infra-provider/api/tidbcloud/cluster-server/v1"
)

const (
	DefaultReplayImage    = "gcr.io/pingcap-public/dbaas/tiproxy-replayer:main-1346b0"
	DefaultReplayReplicas = 1
)

// DefaultReplayOptions returns system level default replay options
func DefaultReplayOptions() *apiv1.Replay_ReplayOptions {
	return &apiv1.Replay_ReplayOptions{
		ExtraArgs: map[string]string{
			"format":       "audit_log_plugin",
			"ignore-errs":  "true",
			"ps-close":     "always",
			"log-file":     "/data/replay-log/replay.log",
			"dynamicinput": "true",
			"bufsize":      "1000000",
			"speed":        "1",
			"wait-on-eof":  "true",
		},
	}
}

// ReplayOptionsBuilder is used to merge replay config by level
// config inheritance priority: system default -> replay level -> single task level -> replay replica level
type ReplayOptionsBuilder struct {
	options cloudv1.ReplayConfig
}

// NewReplayOptionsBuilder creates a config builder, using system default values
func NewReplayOptionsBuilder() *ReplayOptionsBuilder {
	return &ReplayOptionsBuilder{
		options: PB2ReplayOptions(DefaultReplayOptions()),
	}
}

// WithTaskDefaults applies task level default config
func (b *ReplayOptionsBuilder) WithTaskDefaults(defaults *apiv1.Replay_ReplayDefaults) *ReplayOptionsBuilder {
	if defaults == nil {
		return b
	}
	if defaults.Options != nil {
		b.mergeOptions(defaults.Options)
	}
	if defaults.Image != nil && *defaults.Image != "" {
		b.options.Image = *defaults.Image
	}
	return b
}

// WithTaskConfig applies replay task level config
func (b *ReplayOptionsBuilder) WithTaskConfig(task *apiv1.Replay_ReplayTask) *ReplayOptionsBuilder {
	if task == nil {
		return b
	}
	if task.ReplayStartTime != nil {
		b.options.CommandEndTime = *task.ReplayStartTime
	}
	if task.RunStartTime != nil {
		b.options.StartTime = *task.RunStartTime
	}
	if task.Image != nil && *task.Image != "" {
		b.options.Image = *task.Image
	}
	return b
}

// WithReplicaConfig applies replica level config
func (b *ReplayOptionsBuilder) WithReplicaConfig(replicaConfig *apiv1.Replay_ReplicaConfig) *ReplayOptionsBuilder {
	if replicaConfig == nil || replicaConfig.Options == nil {
		return b
	}
	b.mergeOptions(replicaConfig.Options)
	return b
}

// WithReplayerInfo sets replayer distributed info
func (b *ReplayOptionsBuilder) WithReplayerInfo(index, count int32) *ReplayOptionsBuilder {
	b.options.ReplayerIndex = index
	b.options.ReplayerCount = count
	return b
}

// Build returns the final merged config
func (b *ReplayOptionsBuilder) Build() cloudv1.ReplayConfig {
	return b.options
}

// mergeOptions merges ReplayOptions, non-empty fields override
func (b *ReplayOptionsBuilder) mergeOptions(opts *apiv1.Replay_ReplayOptions) {
	if opts == nil {
		return
	}
	if opts.ReplayStartTime != nil {
		b.options.CommandEndTime = *opts.ReplayStartTime
	}
	if opts.RunStartTime != nil {
		b.options.StartTime = *opts.RunStartTime
	}
	maps.Copy(b.options.CustomArguments, opts.ExtraArgs)
}

// ========== ReplayTaskBuilder Replay Group Builder ==========

// ReplayTaskBuilder is used to build replay task config
type ReplayTaskBuilder struct {
	replay *apiv1.Replay
	task   *apiv1.Replay_ReplayTask
}

// NewReplayTaskBuilder creates a replay task builder
func NewReplayTaskBuilder(replay *apiv1.Replay, task *apiv1.Replay_ReplayTask) *ReplayTaskBuilder {
	return &ReplayTaskBuilder{
		replay: replay,
		task:   task,
	}
}

// Build builds cloudv1.ReplayTask
func (b *ReplayTaskBuilder) Build() cloudv1.ReplayTask {
	task := cloudv1.ReplayTask{
		Name:           b.task.Name,
		Replicas:       b.getReplicas(),
		Resources:      b.getResources(),
		TargetEndpoint: b.task.Target.Endpoint,
		Username:       b.getUsername(),
	}

	// build data source
	for _, auditLog := range b.task.Source.AuditLogs {
		task.SourceAuditLogAddrs = append(task.SourceAuditLogAddrs, PB2AuditLogSource(auditLog))
	}

	// build each replicas config
	replicas := b.getReplicas()
	replicaConfigs := b.buildReplicaConfigs()

	for i := range replicas {
		builder := NewReplayOptionsBuilder().
			WithTaskDefaults(b.replay.Defaults).
			WithTaskConfig(b.task).
			WithReplicaConfig(replicaConfigs[i]).
			WithReplayerInfo(i, replicas)

		task.ReplayConfigs = append(task.ReplayConfigs, builder.Build())
	}

	return task
}

// getReplicas gets the number of replicas
func (b *ReplayTaskBuilder) getReplicas() int32 {
	return ReplayTaskReplicas(b.replay, b.task)
}

// getResources gets the resources
func (b *ReplayTaskBuilder) getResources() *cloudv1.ReplayTaskResources {
	return ReplayTaskResources(b.replay, b.task)
}

func (b *ReplayTaskBuilder) getUsername() string {
	return ReplayTaskUsername(b.replay, b.task)
}

func (b *ReplayTaskBuilder) buildReplicaConfigs() map[int32]*apiv1.Replay_ReplicaConfig {
	result := make(map[int32]*apiv1.Replay_ReplicaConfig, len(b.task.ReplicaConfigs))
	for _, config := range b.task.ReplicaConfigs {
		if config == nil {
			continue
		}

		result[config.ReplicaIndex] = config
	}
	return result
}

// PB2ReplayOptions converts proto ReplayOptions to CR ReplayConfig
func PB2ReplayOptions(opts *apiv1.Replay_ReplayOptions) cloudv1.ReplayConfig {
	if opts == nil {
		return cloudv1.ReplayConfig{}
	}
	return cloudv1.ReplayConfig{
		CommandEndTime:  getStringValue(opts.ReplayStartTime),
		StartTime:       getStringValue(opts.RunStartTime),
		CustomArguments: opts.ExtraArgs,
		Image:           DefaultReplayImage,
	}
}

// PB2ReplayResources converts proto ReplayResources to CR ReplayTaskResources
func PB2ReplayResources(res *apiv1.Replay_Resources) *cloudv1.ReplayTaskResources {
	if res == nil {
		return nil
	}
	return &cloudv1.ReplayTaskResources{
		InstanceType: getStringValue(res.InstanceType),
		StorageSize:  getStringValue(res.StorageSize),
	}
}

// PB2AuditLogSource converts proto AuditLogSource to CR AuditLogAddr
func PB2AuditLogSource(source *apiv1.Replay_AuditLogAddr) cloudv1.AuditLogAddr {
	if source == nil {
		return cloudv1.AuditLogAddr{}
	}
	if aws := source.GetAws(); aws != nil {
		return cloudv1.AuditLogAddr{
			AWS: &cloudv1.AWSAuditLog{
				S3URL:      aws.S3Url,
				RoleArn:    aws.RoleArn,
				ExternalID: aws.ExternalId,
			},
		}
	}
	if aliyun := source.GetAliyun(); aliyun != nil {
		return cloudv1.AuditLogAddr{
			Aliyun: &cloudv1.AliyunAuditLog{
				BucketURL: aliyun.BucketUrl,
			},
		}
	}
	return cloudv1.AuditLogAddr{}
}

// PB2ReplayState converts proto ReplayState to CR ReplayState
func PB2ReplayState(state apiv1.Replay_State) cloudv1.ReplayState {
	switch state {
	case apiv1.Replay_STATE_DEPLOY:
		return cloudv1.ReplayStateDeploy
	case apiv1.Replay_STATE_ACTIVE:
		return cloudv1.ReplayStateActive
	case apiv1.Replay_STATE_PAUSE:
		return cloudv1.ReplayStatePause
	default:
		return ""
	}
}

// PB2ClusterType converts proto ReplayClusterType to CR ReplayClusterType
func PB2ClusterType(clusterType apiv1.Replay_ClusterType) cloudv1.ReplayClusterType {
	switch clusterType {
	case apiv1.Replay_CLUSTER_TYPE_DEDICATED:
		return cloudv1.ReplayClusterTypeDedicated
	case apiv1.Replay_CLUSTER_TYPE_ESSENTIAL:
		return cloudv1.ReplayClusterTypeEssential
	case apiv1.Replay_CLUSTER_TYPE_ONPREMISE:
		return cloudv1.ReplayClusterTypeOnPremise
	case apiv1.Replay_CLUSTER_TYPE_PREMIUM:
		return cloudv1.ReplayClusterTypePremium
	case apiv1.Replay_CLUSTER_TYPE_STARTER:
		return cloudv1.ReplayClusterTypeStarter
	default:
		return ""
	}
}

// PB2ReplayTasks converts Tasks in proto Replay to CR ReplayTasks
func PB2ReplayTasks(replay *apiv1.Replay) []cloudv1.ReplayTask {
	if replay == nil || len(replay.Tasks) == 0 {
		return nil
	}

	tasks := make([]cloudv1.ReplayTask, 0, len(replay.Tasks))
	for _, task := range replay.Tasks {
		builder := NewReplayTaskBuilder(replay, task)
		tasks = append(tasks, builder.Build())
	}
	return tasks
}

func ReplayDefaultReplicas(replay *apiv1.Replay) int32 {
	if replay != nil && replay.Defaults != nil && replay.Defaults.Replicas != nil {
		return *replay.Defaults.Replicas
	}
	return DefaultReplayReplicas
}

func ReplayTaskReplicas(replay *apiv1.Replay, task *apiv1.Replay_ReplayTask) int32 {
	if task != nil && task.Replicas != nil {
		return *task.Replicas
	}
	return ReplayDefaultReplicas(replay)
}

func ReplayDefaultResources(replay *apiv1.Replay) *cloudv1.ReplayTaskResources {
	if replay != nil && replay.Defaults != nil && replay.Defaults.Resources != nil {
		return PB2ReplayResources(replay.Defaults.Resources)
	}
	return nil
}

func ReplayTaskResources(replay *apiv1.Replay, task *apiv1.Replay_ReplayTask) *cloudv1.ReplayTaskResources {
	if task != nil && task.Resources != nil {
		return PB2ReplayResources(task.Resources)
	}
	return ReplayDefaultResources(replay)
}

func ReplayDefaultUsername(replay *apiv1.Replay) string {
	if replay != nil && replay.Defaults != nil && replay.Defaults.Credential != nil {
		return replay.Defaults.Credential.Username
	}
	return ""
}

func ReplayTaskUsername(replay *apiv1.Replay, task *apiv1.Replay_ReplayTask) string {
	if task != nil && task.Target != nil && task.Target.Credential != nil {
		return task.Target.Credential.Username
	}
	return ReplayDefaultUsername(replay)
}

func ReplayDefaultReplayStartTime(replay *apiv1.Replay) string {
	if replay != nil && replay.Defaults != nil && replay.Defaults.Options != nil && replay.Defaults.Options.ReplayStartTime != nil {
		return *replay.Defaults.Options.ReplayStartTime
	}
	return ""
}

func ReplayDefaultRunStartTime(replay *apiv1.Replay) string {
	if replay != nil && replay.Defaults != nil && replay.Defaults.Options != nil && replay.Defaults.Options.RunStartTime != nil {
		return *replay.Defaults.Options.RunStartTime
	}
	return ""
}

// Replay2PB converts CR Replay to proto Replay
func Replay2PB(replay *cloudv1.Replay) *apiv1.Replay {
	if replay == nil {
		return nil
	}

	out := &apiv1.Replay{
		ReplayId:  replay.Name,
		TenantId:  replay.GetTenantID(),
		ProjectId: replay.GetProjectID(),
		State:     ReplayState2PB(replay.Spec.State),
		TargetCluster: &apiv1.Replay_ClusterRef{
			ClusterType: ClusterType2PB(replay.Spec.TargetClusterType),
			ClusterId:   replay.Spec.TargetClusterID,
			Provider:    replay.Spec.TargetProvider,
			Region:      replay.Spec.TargetRegion,
		},
		DataplaneKubernetes: &apiv1.Kubernetes{
			Name: replay.Spec.Kubernetes.Name,
		},
	}

	if replay.Spec.SourceClusterType != nil {
		out.SourceCluster = &apiv1.Replay_ClusterRef{
			ClusterType: ClusterType2PB(*replay.Spec.SourceClusterType),
			ClusterId:   Coalesce(replay.Spec.SourceClusterID, ""),
			Provider:    Coalesce(replay.Spec.SourceProvider, ""),
			Region:      Coalesce(replay.Spec.SourceRegion, ""),
		}
	}

	return out
}

// ReplayStatus2PB converts CR ReplayStatus to proto ReplayStatus
func ReplayStatus2PB(status *cloudv1.ReplayStatus) *apiv1.ReplayStatus {
	if status == nil {
		return nil
	}
	return &apiv1.ReplayStatus{
		Phase:          ReplayPhase2PB(status.Phase),
		Message:        status.Message,
		Tasks:          BatchReplayTaskStatus2PB(status.Tasks),
		LastUpdateTime: ToGRPCTimestamp(status.LastUpdateTime),
		AwsNodeRoleArn: status.AWSNodeRoleARN,
	}
}

// ReplayState2PB converts CR ReplayState to proto ReplayState
func ReplayState2PB(state cloudv1.ReplayState) apiv1.Replay_State {
	switch state {
	case cloudv1.ReplayStateDeploy:
		return apiv1.Replay_STATE_DEPLOY
	case cloudv1.ReplayStateActive:
		return apiv1.Replay_STATE_ACTIVE
	case cloudv1.ReplayStatePause:
		return apiv1.Replay_STATE_PAUSE
	default:
		return apiv1.Replay_STATE_UNSPECIFIED
	}
}

// ClusterType2PB converts CR ReplayClusterType to proto ReplayClusterType
func ClusterType2PB(clusterType cloudv1.ReplayClusterType) apiv1.Replay_ClusterType {
	switch clusterType {
	case cloudv1.ReplayClusterTypeDedicated:
		return apiv1.Replay_CLUSTER_TYPE_DEDICATED
	case cloudv1.ReplayClusterTypeEssential:
		return apiv1.Replay_CLUSTER_TYPE_ESSENTIAL
	case cloudv1.ReplayClusterTypeOnPremise:
		return apiv1.Replay_CLUSTER_TYPE_ONPREMISE
	case cloudv1.ReplayClusterTypePremium:
		return apiv1.Replay_CLUSTER_TYPE_PREMIUM
	case cloudv1.ReplayClusterTypeStarter:
		return apiv1.Replay_CLUSTER_TYPE_STARTER
	default:
		return apiv1.Replay_CLUSTER_TYPE_UNSPECIFIED
	}
}

// ReplayPhase2PB converts CR ReplayPhase to proto ReplayPhase
func ReplayPhase2PB(phase cloudv1.ReplayPhase) apiv1.ReplayStatus_Phase {
	switch phase {
	case cloudv1.ReplayPending:
		return apiv1.ReplayStatus_PHASE_PENDING
	case cloudv1.ReplayCreating:
		return apiv1.ReplayStatus_PHASE_CREATING
	case cloudv1.ReplayRunning:
		return apiv1.ReplayStatus_PHASE_RUNNING
	case cloudv1.ReplayStopped:
		return apiv1.ReplayStatus_PHASE_STOPPED
	case cloudv1.ReplayDeleting:
		return apiv1.ReplayStatus_PHASE_DELETING
	default:
		return apiv1.ReplayStatus_PHASE_UNSPECIFIED
	}
}

// ReplayTaskPhase2PB converts CR ReplayTaskPhase to proto ReplayTaskPhase
func ReplayTaskPhase2PB(phase cloudv1.ReplayTaskPhase) apiv1.ReplayStatus_TaskPhase {
	switch phase {
	case cloudv1.ReplayTaskPhasePending:
		return apiv1.ReplayStatus_TASK_PHASE_PENDING
	case cloudv1.ReplayTaskPhaseRunning:
		return apiv1.ReplayStatus_TASK_PHASE_RUNNING
	case cloudv1.ReplayTaskPhaseStopped:
		return apiv1.ReplayStatus_TASK_PHASE_STOPPED
	case cloudv1.ReplayTaskPhaseDeleted:
		return apiv1.ReplayStatus_TASK_PHASE_DELETED
	default:
		return apiv1.ReplayStatus_TASK_PHASE_UNSPECIFIED
	}
}

// BatchReplayTaskStatus2PB converts batch of CR ReplayTaskStatus to proto ReplayStatus_ReplayTaskStatus
func BatchReplayTaskStatus2PB(crTasks []cloudv1.ReplayTaskStatus) []*apiv1.ReplayStatus_ReplayTaskStatus {
	if len(crTasks) == 0 {
		return nil
	}

	grpcTasks := make([]*apiv1.ReplayStatus_ReplayTaskStatus, 0, len(crTasks))
	for _, crTask := range crTasks {
		grpcTasks = append(grpcTasks, ReplayTaskStatus2PB(crTask))
	}
	return grpcTasks
}

// ReplayTaskStatus2PB converts CR ReplayTaskStatus to proto ReplayTaskStatus
func ReplayTaskStatus2PB(crTask cloudv1.ReplayTaskStatus) *apiv1.ReplayStatus_ReplayTaskStatus {
	return &apiv1.ReplayStatus_ReplayTaskStatus{
		Name:            crTask.Name,
		ReplicaStatuses: ReplayReplicaStatuses2PB(crTask.ReplicaStatuses),
		Message:         crTask.Message,
	}
}

// ReplayReplicaStatuses2PB converts batch of CR ReplayReplicaStatus to proto ReplayStatus_ReplayReplicaStatus
func ReplayReplicaStatuses2PB(crReplicas []cloudv1.ReplayReplicaStatus) []*apiv1.ReplayStatus_ReplayReplicaStatus {
	if len(crReplicas) == 0 {
		return nil
	}

	grpcReplicas := make([]*apiv1.ReplayStatus_ReplayReplicaStatus, 0, len(crReplicas))
	for _, crReplica := range crReplicas {
		grpcReplicas = append(grpcReplicas, ReplayReplicaStatus2PB(crReplica))
	}
	return grpcReplicas
}

// ReplayReplicaStatus2PB converts CR ReplayReplicaStatus to proto ReplayReplicaStatus
func ReplayReplicaStatus2PB(crReplica cloudv1.ReplayReplicaStatus) *apiv1.ReplayStatus_ReplayReplicaStatus {
	return &apiv1.ReplayStatus_ReplayReplicaStatus{
		PodName:  crReplica.PodName,
		Index:    crReplica.Index,
		Phase:    ReplayTaskPhase2PB(crReplica.Phase),
		Ready:    crReplica.Ready,
		Message:  crReplica.Message,
		TaskInfo: ReplayTaskInfo2PB(crReplica.TaskInfo),
	}
}

// ReplayTaskInfo2PB converts CR ReplayTaskInfo to proto ReplayTaskInfo
func ReplayTaskInfo2PB(crTaskInfo *cloudv1.ReplayTaskInfo) *apiv1.ReplayStatus_ReplayTaskInfo {
	if crTaskInfo == nil {
		return nil
	}

	return &apiv1.ReplayStatus_ReplayTaskInfo{
		Type:                 crTaskInfo.Type,
		Status:               crTaskInfo.Status,
		StartTime:            crTaskInfo.StartTime,
		EndTime:              crTaskInfo.EndTime,
		Progress:             crTaskInfo.Progress,
		Error:                crTaskInfo.Error,
		LastCommandTimestamp: crTaskInfo.LastCommandTimestamp,
		SourceAuditLogAddrs:  AuditLogAddrs2PB(crTaskInfo.SourceAuditLogAddrs),
		TotalCommands:        crTaskInfo.TotalCommands,
		ReplayedCommands:     crTaskInfo.ReplayedCommands,
		FailedCommands:       crTaskInfo.FailedCommands,
		TotalBytes:           crTaskInfo.TotalBytes,
		ProcessedBytes:       crTaskInfo.ProcessedBytes,
	}
}

// AuditLogAddrs2PB converts list of CR AuditLogAddr to list of proto AuditLogAddr
func AuditLogAddrs2PB(crAddrs []cloudv1.AuditLogAddr) []*apiv1.Replay_AuditLogAddr {
	if len(crAddrs) == 0 {
		return nil
	}

	grpcAddrs := make([]*apiv1.Replay_AuditLogAddr, 0, len(crAddrs))
	for _, addr := range crAddrs {
		pbAddr := &apiv1.Replay_AuditLogAddr{}
		if addr.AWS != nil {
			pbAddr.Source = &apiv1.Replay_AuditLogAddr_Aws{
				Aws: &apiv1.Replay_AWSAuditLog{
					S3Url:      addr.AWS.S3URL,
					RoleArn:    addr.AWS.RoleArn,
					ExternalId: addr.AWS.ExternalID,
				},
			}
		}
		if addr.Aliyun != nil {
			pbAddr.Source = &apiv1.Replay_AuditLogAddr_Aliyun{
				Aliyun: &apiv1.Replay_AliyunAuditLog{
					BucketUrl: addr.Aliyun.BucketURL,
				},
			}
		}
		grpcAddrs = append(grpcAddrs, pbAddr)
	}
	return grpcAddrs
}

// ToGRPCTimestamp converts metav1.Time to proto Timestamp
func ToGRPCTimestamp(metav1Time metav1.Time) *timestamppb.Timestamp {
	if metav1Time.IsZero() {
		return nil
	}
	return timestamppb.New(metav1Time.Time)
}

// Coalesce returns the value pointed to by the pointer, if the pointer is nil, returns the default value
func Coalesce[T any](ptr *T, defaultValue T) T {
	if ptr != nil {
		return *ptr
	}
	return defaultValue
}

func getStringValue(ptr *string) string {
	if ptr != nil {
		return *ptr
	}
	return ""
}