package inner

import (
	"context"

	"github.com/tidbcloud/dataflow-service/app/models/domain"
	"github.com/tidbcloud/dataflow-service/app/models/replay"
	"github.com/tidbcloud/dataflow-service/app/repo/entity"
	pbReplay "github.com/tidbcloud/dataflow-service/idl/pbgen/proto/replay"
	"github.com/tidbcloud/dataflow-service/internal/utils"
	"github.com/tidbcloud/dataflow-service/pkg/log"
	"github.com/tidbcloud/pkgv2/bizerr"
	"go.uber.org/zap"
	"google.golang.org/protobuf/types/known/emptypb"
	"google.golang.org/protobuf/types/known/timestamppb"
)

type ReplayInnerController interface {
	pbReplay.ReplayServiceServer
}

type replayInnerControllerImpl struct {
	pbReplay.UnimplementedReplayServiceServer

	replayModel replay.ReplayModel
}

func NewReplayInnerController(
	replayModel replay.ReplayModel,
) ReplayInnerController {
	return &replayInnerControllerImpl{
		replayModel: replayModel,
	}
}

func (c *replayInnerControllerImpl) CreateReplay(
	ctx context.Context, req *pbReplay.CreateReplayRequest,
) (*pbReplay.CreateReplayResponse, error) {
	ctx = log.WithTraceID(ctx)
	logger := log.GetLogger(ctx).With(
		zap.String("org_id", req.OrgId),
		zap.String("cluster_id", req.ClusterId),
	)

	// Validate request
	if err := c.validateCreateReplayRequest(req); err != nil {
		logger.Warn("invalid create replay request", zap.Error(err))
		return nil, err
	}

	// Convert protobuf request to domain request
	createReq, err := c.pbReplaySpec2DomainCreateReq(req)
	if err != nil {
		logger.Error("convert pb replay spec to domain create req error", zap.Error(err))
		return nil, err
	}

	// Create replay
	createResp, err := c.replayModel.Create(ctx, createReq)
	if err != nil {
		logger.Error("create replay error", zap.Error(err))
		return nil, err
	}

	// Get created replay for response
	getResp, err := c.replayModel.Get(ctx, &domain.GetReplayReq{
		ReplayID: createResp.ReplayID,
	})
	if err != nil {
		logger.Error("get replay after creation error",
			zap.Uint64("replay_id", createResp.ReplayID),
			zap.Error(err))
		return nil, err
	}

	logger.Info("create replay success", zap.Uint64("replay_id", createResp.ReplayID))
	return &pbReplay.CreateReplayResponse{
		ReplayId: utils.Uint64ToString(createResp.ReplayID),
		Replay:   c.domainReplay2PB(getResp.Replay),
	}, nil
}

func (c *replayInnerControllerImpl) ListReplays(
	ctx context.Context, req *pbReplay.ListReplaysRequest,
) (*pbReplay.ListReplaysResponse, error) {
	ctx = log.WithTraceID(ctx)
	logger := log.GetLogger(ctx).With(zap.String("org_id", req.OrgId))

	// Validate and convert org_id
	tenantID, err := utils.StringToUint64(req.OrgId)
	if err != nil {
		logger.Warn("invalid org_id", zap.String("org_id", req.OrgId), zap.Error(err))
		return nil, bizerr.New(bizerr.ErrCodeParams, "invalid org_id: %s", req.OrgId)
	}

	// Convert request
	listReq := &domain.ListReplayReq{
		TenantID: tenantID,
		Page:     uint32(req.Page),
		PageSize: uint32(req.PageSize),
	}

	if req.ClusterId != "" {
		clusterID, err := utils.StringToUint64(req.ClusterId)
		if err != nil {
			logger.Warn("invalid cluster_id", zap.String("cluster_id", req.ClusterId), zap.Error(err))
			return nil, bizerr.New(bizerr.ErrCodeParams, "invalid cluster_id: %s", req.ClusterId)
		}
		listReq.ClusterID = clusterID
	}

	// Set default page size
	if listReq.PageSize == 0 {
		listReq.PageSize = 20
	}

	// List replays
	listResp, bizErr := c.replayModel.List(ctx, listReq)
	if bizErr != nil {
		logger.Error("list replays error", zap.Error(bizErr))
		return nil, bizErr
	}

	// Convert to protobuf
	pbReplays := make([]*pbReplay.Replay, 0, len(listResp.Replays))
	for _, r := range listResp.Replays {
		pbReplays = append(pbReplays, c.domainReplay2PB(r))
	}

	logger.Info("list replays success", zap.Int("count", len(pbReplays)))
	return &pbReplay.ListReplaysResponse{
		Replays:   pbReplays,
		TotalSize: int32(listResp.Total),
	}, nil
}

func (c *replayInnerControllerImpl) GetReplay(
	ctx context.Context, req *pbReplay.GetReplayRequest,
) (*pbReplay.GetReplayResponse, error) {
	ctx = log.WithTraceID(ctx)
	logger := log.GetLogger(ctx).With(
		zap.String("org_id", req.OrgId),
		zap.String("replay_id", req.ReplayId),
	)

	// Validate replay_id
	replayID, err := utils.StringToUint64(req.ReplayId)
	if err != nil {
		logger.Warn("invalid replay_id", zap.String("replay_id", req.ReplayId), zap.Error(err))
		return nil, bizerr.New(bizerr.ErrCodeParams, "invalid replay_id: %s", req.ReplayId)
	}

	// Get replay
	getResp, bizErr := c.replayModel.Get(ctx, &domain.GetReplayReq{
		ReplayID: replayID,
	})
	if bizErr != nil {
		logger.Error("get replay error", zap.Error(bizErr))
		return nil, bizErr
	}

	logger.Info("get replay success")
	return &pbReplay.GetReplayResponse{
		Replay: c.domainReplay2PB(getResp.Replay),
	}, nil
}

func (c *replayInnerControllerImpl) UpdateReplay(
	ctx context.Context, req *pbReplay.UpdateReplayRequest,
) (*pbReplay.UpdateReplayResponse, error) {
	ctx = log.WithTraceID(ctx)
	logger := log.GetLogger(ctx).With(
		zap.String("org_id", req.OrgId),
		zap.String("replay_id", req.ReplayId),
	)

	// Validate request
	if err := c.validateUpdateReplayRequest(req); err != nil {
		logger.Warn("invalid update replay request", zap.Error(err))
		return nil, err
	}

	// Convert protobuf request to domain request
	updateReq, err := c.pbReplaySpec2DomainUpdateReq(req)
	if err != nil {
		logger.Error("convert pb replay spec to domain update req error", zap.Error(err))
		return nil, err
	}

	// Update replay
	updateResp, err := c.replayModel.Update(ctx, updateReq)
	if err != nil {
		logger.Error("update replay error", zap.Error(err))
		return nil, err
	}

	logger.Info("update replay success")
	return &pbReplay.UpdateReplayResponse{
		Replay: c.domainReplay2PB(updateResp.Replay),
	}, nil
}

//nolint:dupl
func (c *replayInnerControllerImpl) DeleteReplay(
	ctx context.Context, req *pbReplay.DeleteReplayRequest,
) (*emptypb.Empty, error) {
	ctx = log.WithTraceID(ctx)
	logger := log.GetLogger(ctx).With(
		zap.String("org_id", req.OrgId),
		zap.String("replay_id", req.ReplayId),
	)

	// Validate and convert replay_id
	replayID, err := utils.StringToUint64(req.ReplayId)
	if err != nil {
		logger.Warn("invalid replay_id", zap.String("replay_id", req.ReplayId), zap.Error(err))
		return nil, bizerr.New(bizerr.ErrCodeParams, "invalid replay_id: %s", req.ReplayId)
	}

	// Delete replay
	bizErr := c.replayModel.Delete(ctx, &domain.DeleteReplayReq{
		ReplayID: replayID,
	})
	if bizErr != nil {
		logger.Error("delete replay error", zap.Error(bizErr))
		return nil, bizErr
	}

	logger.Info("delete replay success")
	return &emptypb.Empty{}, nil
}

//nolint:dupl
func (c *replayInnerControllerImpl) StartReplay(
	ctx context.Context, req *pbReplay.StartReplayRequest,
) (*emptypb.Empty, error) {
	ctx = log.WithTraceID(ctx)
	logger := log.GetLogger(ctx).With(
		zap.String("org_id", req.OrgId),
		zap.String("replay_id", req.ReplayId),
	)

	// Validate and convert replay_id
	replayID, err := utils.StringToUint64(req.ReplayId)
	if err != nil {
		logger.Warn("invalid replay_id", zap.String("replay_id", req.ReplayId), zap.Error(err))
		return nil, bizerr.New(bizerr.ErrCodeParams, "invalid replay_id: %s", req.ReplayId)
	}

	// Resume replay
	bizErr := c.replayModel.Resume(ctx, &domain.ResumeReplayReq{
		ReplayID: replayID,
	})
	if bizErr != nil {
		logger.Error("start replay error", zap.Error(bizErr))
		return nil, bizErr
	}

	logger.Info("start replay success")
	return &emptypb.Empty{}, nil
}

//nolint:dupl
func (c *replayInnerControllerImpl) StopReplay(
	ctx context.Context, req *pbReplay.StopReplayRequest,
) (*emptypb.Empty, error) {
	ctx = log.WithTraceID(ctx)
	logger := log.GetLogger(ctx).With(
		zap.String("org_id", req.OrgId),
		zap.String("replay_id", req.ReplayId),
	)

	// Validate and convert replay_id
	replayID, err := utils.StringToUint64(req.ReplayId)
	if err != nil {
		logger.Warn("invalid replay_id", zap.String("replay_id", req.ReplayId), zap.Error(err))
		return nil, bizerr.New(bizerr.ErrCodeParams, "invalid replay_id: %s", req.ReplayId)
	}

	// Pause replay
	bizErr := c.replayModel.Pause(ctx, &domain.PauseReplayReq{
		ReplayID: replayID,
	})
	if bizErr != nil {
		logger.Error("stop replay error", zap.Error(bizErr))
		return nil, bizErr
	}

	logger.Info("stop replay success")
	return &emptypb.Empty{}, nil
}

func (c *replayInnerControllerImpl) pbReplaySpec2DomainCreateReq(
	req *pbReplay.CreateReplayRequest,
) (*domain.CreateReplayReq, bizerr.BizError) {
	spec := req.Replay
	if spec == nil {
		return nil, bizerr.New(bizerr.ErrCodeParams, "replay spec is required")
	}

	tenantID, err := utils.StringToUint64(req.OrgId)
	if err != nil {
		return nil, bizerr.New(bizerr.ErrCodeParams, "invalid org_id: %s", req.OrgId)
	}

	projectID, err := utils.StringToUint64(spec.ProjectId)
	if err != nil {
		return nil, bizerr.New(bizerr.ErrCodeParams, "invalid org_id: %s", req.OrgId)
	}

	sourceClusterID, err := utils.StringToUint64(spec.SourceClusterId)
	if err != nil {
		return nil, bizerr.New(bizerr.ErrCodeParams, "invalid source_cluster_id: %s", spec.SourceClusterId)
	}

	targetClusterID, err := utils.StringToUint64(spec.TargetClusterId)
	if err != nil {
		return nil, bizerr.New(bizerr.ErrCodeParams, "invalid target_cluster_id: %s", spec.TargetClusterId)
	}

	createReq := &domain.CreateReplayReq{
		TenantID:        tenantID,
		ProjectID:       projectID,
		CloudProvider:   spec.CloudProvider,
		Region:          spec.Region,
		SourceClusterID: sourceClusterID,
		TargetClusterID: targetClusterID,
		Replicas:        spec.Replicas,
		ReplayImage:     spec.ReplayImage,
	}

	// Convert replay tasks
	if len(spec.ReplayTasks) > 0 {
		createReq.ReplayTasks = c.pbReplayTasks2Entity(spec.ReplayTasks)
	}

	// Convert replay config
	if spec.ReplayConfig != nil {
		createReq.ReplayConfig = c.pbReplayCommonConfig2Entity(spec.ReplayConfig)
	}

	// Convert task resources
	if spec.TaskResources != nil {
		createReq.TaskResources = c.pbTaskResources2Entity(spec.TaskResources)
	}

	// Convert source audit credential
	if spec.SourceAuditCredential != nil {
		addr := c.pbAuditLogAddr2Entity(spec.SourceAuditCredential)
		createReq.SourceAuditCredential = &addr
	}

	return createReq, nil
}

func (c *replayInnerControllerImpl) pbReplaySpec2DomainUpdateReq(
	req *pbReplay.UpdateReplayRequest,
) (*domain.UpdateReplayReq, bizerr.BizError) {
	spec := req.Replay
	replayID, err := utils.StringToUint64(req.ReplayId)
	if err != nil {
		return nil, bizerr.New(bizerr.ErrCodeParams, "invalid replay_id: %s", req.ReplayId)
	}
	updateReq := &domain.UpdateReplayReq{
		ReplayID: replayID,
	}
	if spec.CloudProvider != "" {
		updateReq.CloudProvider = &spec.CloudProvider
	}
	if spec.Region != "" {
		updateReq.Region = &spec.Region
	}
	if spec.SourceClusterId != "" {
		sourceClusterID, err := utils.StringToUint64(spec.SourceClusterId)
		if err != nil {
			return nil, bizerr.New(bizerr.ErrCodeParams, "invalid source_cluster_id: %s", spec.SourceClusterId)
		}
		updateReq.SourceClusterID = &sourceClusterID
	}
	if spec.TargetClusterId != "" {
		targetClusterID, err := utils.StringToUint64(spec.TargetClusterId)
		if err != nil {
			return nil, bizerr.New(bizerr.ErrCodeParams, "invalid target_cluster_id: %s", spec.TargetClusterId)
		}
		updateReq.TargetClusterID = &targetClusterID
	}
	if len(spec.ReplayTasks) > 0 {
		updateReq.ReplayTasks = c.pbReplayTasks2Entity(spec.ReplayTasks)
	}
	if spec.ReplayConfig != nil {
		updateReq.ReplayConfig = c.pbReplayCommonConfig2Entity(spec.ReplayConfig)
	}
	if spec.TaskResources != nil {
		updateReq.TaskResources = c.pbTaskResources2Entity(spec.TaskResources)
	}
	if spec.ReplayImage != "" {
		updateReq.ReplayImage = &spec.ReplayImage
	}
	if spec.Replicas != 0 {
		updateReq.Replicas = &spec.Replicas
	}
	if spec.SourceAuditCredential != nil {
		addr := c.pbAuditLogAddr2Entity(spec.SourceAuditCredential)
		updateReq.SourceAuditCredential = &addr
	}

	return updateReq, nil
}

func (c *replayInnerControllerImpl) domainReplay2PB(r *domain.Replay) *pbReplay.Replay {
	if r == nil {
		return nil
	}

	pbR := &pbReplay.Replay{
		ReplayId:        utils.Uint64ToString(r.ID),
		OrgId:           utils.Uint64ToString(r.TenantID),
		ProjectId:       utils.Uint64ToString(r.ProjectID),
		ClusterId:       utils.Uint64ToString(r.TargetClusterID),
		DesiredState:    c.entityDesiredState2PB(r.DesiredState),
		CurrentState:    c.entityCurrentState2PB(r.CurrentState),
		SourceClusterId: utils.Uint64ToString(r.SourceClusterID),
		CloudProvider:   r.CloudProvider,
		Region:          r.Region,
		Replicas:        r.Replicas,
		ReplayImage:     r.ReplayImage,
		CreatedAt:       timestamppb.New(r.CreatedAt),
	}

	if r.UpdatedAt != nil {
		pbR.UpdatedAt = timestamppb.New(*r.UpdatedAt)
	}

	if len(r.ReplayTasks) > 0 {
		pbR.ReplayTasks = c.entityReplayTasks2PB(r.ReplayTasks)
	}

	if r.ReplayConfig != nil {
		pbR.ReplayConfig = c.entityReplayCommonConfig2PB(r.ReplayConfig)
	}

	if r.TaskResources != nil {
		pbR.TaskResources = c.entityTaskResources2PB(r.TaskResources)
	}

	if r.ReplayStatus != nil {
		pbR.ReplayStatus = c.entityReplayStatus2PB(r.ReplayStatus)
	}

	if r.SourceAuditCredential != nil {
		pbR.SourceAuditCredential = c.entityAuditLogAddr2PB(*r.SourceAuditCredential)
	}

	return pbR
}

func (c *replayInnerControllerImpl) pbReplayTasks2Entity(tasks []*pbReplay.ReplayTask) entity.ReplayTaskList {
	result := make(entity.ReplayTaskList, 0, len(tasks))
	for _, t := range tasks {
		result = append(result, c.pbReplayTask2Entity(t))
	}
	return result
}

func (c *replayInnerControllerImpl) pbReplayTask2Entity(t *pbReplay.ReplayTask) entity.ReplayTask {
	task := entity.ReplayTask{
		Name:            t.Name,
		Replicas:        t.Replicas,
		ReplayStartTime: t.ReplayStartTime,
		RunStartTime:    t.RunStartTime,
		Image:           t.Image,
	}

	if t.DataSource != nil {
		task.DataSource = c.pbDataSource2Entity(t.DataSource)
	}

	if t.DataTarget != nil {
		task.DataTarget = c.pbDataTarget2Entity(t.DataTarget)
	}

	if t.Resources != nil {
		task.Resources = c.pbTaskResources2Entity(t.Resources)
	}

	if len(t.PerReplayConfigs) > 0 {
		task.PerReplayConfigs = c.pbReplayConfigs2Entity(t.PerReplayConfigs)
	}

	return task
}

func (c *replayInnerControllerImpl) pbDataSource2Entity(ds *pbReplay.ReplayDataSource) entity.ReplayDataSource {
	result := entity.ReplayDataSource{}
	if len(ds.SourceAuditLogAddrs) > 0 {
		result.SourceAuditLogAddrs = c.pbAuditLogAddrs2Entity(ds.SourceAuditLogAddrs)
	}
	return result
}

func (c *replayInnerControllerImpl) pbDataTarget2Entity(dt *pbReplay.ReplayDataTarget) entity.ReplayDataTarget {
	result := entity.ReplayDataTarget{
		Endpoint: dt.Endpoint,
	}
	if dt.Username != "" {
		result.Username = utils.ToPtr(dt.Username)
	}
	if dt.Password != "" {
		result.Password = utils.ToPtr(dt.Password)
	}
	return result
}

func (c *replayInnerControllerImpl) pbAuditLogAddrs2Entity(addrs []*pbReplay.AuditLogAddr) []entity.AuditLogAddr {
	result := make([]entity.AuditLogAddr, 0, len(addrs))
	for _, addr := range addrs {
		result = append(result, c.pbAuditLogAddr2Entity(addr))
	}
	return result
}

func (c *replayInnerControllerImpl) pbAuditLogAddr2Entity(addr *pbReplay.AuditLogAddr) entity.AuditLogAddr {
	result := entity.AuditLogAddr{}
	if aws := addr.GetAws(); aws != nil {
		result.AWS = &entity.AWSAuditLog{
			S3URL:      aws.S3Url,
			RoleARN:    aws.RoleArn,
			ExternalID: utils.ToPtr(aws.ExternalId),
		}
	}
	if gcp := addr.GetGcp(); gcp != nil {
		result.GCP = &entity.GCPAuditLog{
			GCSURL:             gcp.GcsUrl,
			ServiceAccountJSON: utils.ToPtr(gcp.ServiceAccountJson),
		}
	}
	if aliyun := addr.GetAliyun(); aliyun != nil {
		result.Aliyun = &entity.AliyunAuditLog{
			BucketURL:       aliyun.BucketUrl,
			AccessKeyID:     aliyun.AccessKeyId,
			SecretAccessKey: aliyun.SecretAccessKey,
		}
	}
	return result
}

func (c *replayInnerControllerImpl) pbReplayCommonConfig2Entity(
	cfg *pbReplay.ReplayCommonConfig,
) *entity.ReplayCommonConfig {
	if cfg == nil {
		return nil
	}

	result := &entity.ReplayCommonConfig{}

	if cfg.Username != "" {
		result.Username = cfg.Username
	}
	if cfg.Password != "" {
		result.Password = cfg.Password
	}
	if cfg.ReplayConfig != nil {
		result.ReplayConfig = c.pbReplayConfig2Entity(cfg.ReplayConfig)
	}

	return result
}

func (c *replayInnerControllerImpl) pbReplayConfig2Entity(
	cfg *pbReplay.ReplayConfig,
) entity.ReplayConfig {
	return entity.ReplayConfig{
		ReplayStartTime: utils.ToPtr(cfg.ReplayStartTime),
		RunStartTime:    utils.ToPtr(cfg.RunStartTime),
		ReplayIndex:     utils.ToPtr(cfg.ReplayIndex),
		Format:          utils.ToPtr(cfg.Format),
		IgnoreErrs:      utils.ToPtr(cfg.IgnoreErrs),
		PsClose:         utils.ToPtr(cfg.PsClose),
		BufSize:         utils.ToPtr(cfg.BufSize),
		Speed:           utils.ToPtr(cfg.Speed),
		DynamicInput:    utils.ToPtr(cfg.DynamicInput),
		ReadOnly:        utils.ToPtr(cfg.ReadOnly),
		CustomArguments: cfg.CustomArguments,
	}
}

func (c *replayInnerControllerImpl) pbReplayConfigs2Entity(cfgs []*pbReplay.ReplayConfig) []entity.ReplayConfig {
	result := make([]entity.ReplayConfig, 0, len(cfgs))
	for _, cfg := range cfgs {
		result = append(result, c.pbReplayConfig2Entity(cfg))
	}
	return result
}

func (c *replayInnerControllerImpl) pbTaskResources2Entity(
	res *pbReplay.ReplayTaskResources,
) *entity.ReplayTaskResources {
	if res == nil {
		return nil
	}
	return &entity.ReplayTaskResources{
		InstanceType: res.InstanceType,
		StorageSize:  res.StorageSize,
	}
}

func (c *replayInnerControllerImpl) entityReplayTasks2PB(tasks entity.ReplayTaskList) []*pbReplay.ReplayTask {
	result := make([]*pbReplay.ReplayTask, 0, len(tasks))
	for _, t := range tasks {
		result = append(result, c.entityReplayTask2PB(t))
	}
	return result
}

func (c *replayInnerControllerImpl) entityReplayTask2PB(t entity.ReplayTask) *pbReplay.ReplayTask {
	return &pbReplay.ReplayTask{
		Name:             t.Name,
		DataSource:       c.entityDataSource2PB(t.DataSource),
		DataTarget:       c.entityDataTarget2PB(t.DataTarget),
		Replicas:         t.Replicas,
		ReplayStartTime:  t.ReplayStartTime,
		RunStartTime:     t.RunStartTime,
		Resources:        c.entityTaskResources2PB(t.Resources),
		PerReplayConfigs: c.entityReplayConfigs2PB(t.PerReplayConfigs),
		Image:            t.Image,
	}
}

func (c *replayInnerControllerImpl) entityDataSource2PB(ds entity.ReplayDataSource) *pbReplay.ReplayDataSource {
	return &pbReplay.ReplayDataSource{
		SourceAuditLogAddrs: c.entityAuditLogAddrs2PB(ds.SourceAuditLogAddrs),
	}
}

const (
	passwordPlaceholder        = "{PASSWORD}"
	accessKeyIDPlaceholder     = "{ACCESS_KEY_ID}"
	secretAccessKeyPlaceholder = "{SECRET_ACCESS_KEY}"
)

func (c *replayInnerControllerImpl) entityDataTarget2PB(dt entity.ReplayDataTarget) *pbReplay.ReplayDataTarget {
	password := ""
	if dt.Password != nil && *dt.Password != "" {
		password = passwordPlaceholder
	}
	return &pbReplay.ReplayDataTarget{
		Endpoint: dt.Endpoint,
		Username: utils.PtrToString(dt.Username),
		Password: password,
	}
}

func (c *replayInnerControllerImpl) entityAuditLogAddrs2PB(addrs []entity.AuditLogAddr) []*pbReplay.AuditLogAddr {
	result := make([]*pbReplay.AuditLogAddr, 0, len(addrs))
	for _, addr := range addrs {
		result = append(result, c.entityAuditLogAddr2PB(addr))
	}
	return result
}

func (c *replayInnerControllerImpl) entityAuditLogAddr2PB(addr entity.AuditLogAddr) *pbReplay.AuditLogAddr {
	result := &pbReplay.AuditLogAddr{}
	if addr.AWS != nil {
		result.Storage = &pbReplay.AuditLogAddr_Aws{
			Aws: &pbReplay.AWSAuditLog{
				S3Url:      addr.AWS.S3URL,
				RoleArn:    addr.AWS.RoleARN,
				ExternalId: utils.PtrToString(addr.AWS.ExternalID),
			},
		}
	}
	if addr.GCP != nil {
		result.Storage = &pbReplay.AuditLogAddr_Gcp{
			Gcp: &pbReplay.GCPAuditLog{
				GcsUrl:             addr.GCP.GCSURL,
				ServiceAccountJson: utils.PtrToString(addr.GCP.ServiceAccountJSON),
			},
		}
	}
	if addr.Aliyun != nil {
		result.Storage = &pbReplay.AuditLogAddr_Aliyun{
			Aliyun: &pbReplay.AliyunAuditLog{
				BucketUrl:       domain.RedactAliyunBucketURLForTransport(addr.Aliyun.BucketURL),
				AccessKeyId:     secretPlaceholderForPB(addr.Aliyun.AccessKeyID, accessKeyIDPlaceholder),
				SecretAccessKey: secretPlaceholderForPB(addr.Aliyun.SecretAccessKey, secretAccessKeyPlaceholder),
			},
		}
	}
	return result
}

func secretPlaceholderForPB(secret string, placeholder string) string {
	if secret == "" {
		return ""
	}
	return placeholder
}

func (c *replayInnerControllerImpl) entityReplayCommonConfig2PB(
	cfg *entity.ReplayCommonConfig,
) *pbReplay.ReplayCommonConfig {
	if cfg == nil {
		return nil
	}
	password := ""
	if cfg.Password != "" {
		password = passwordPlaceholder
	}
	return &pbReplay.ReplayCommonConfig{
		ReplayConfig: c.entityReplayConfig2PB(cfg.ReplayConfig),
		Username:     cfg.Username,
		Password:     password,
	}
}

func (c *replayInnerControllerImpl) entityReplayConfig2PB(cfg entity.ReplayConfig) *pbReplay.ReplayConfig {
	return &pbReplay.ReplayConfig{
		ReplayStartTime: utils.PtrToString(cfg.ReplayStartTime),
		RunStartTime:    utils.PtrToString(cfg.RunStartTime),
		ReplayIndex:     utils.PtrToInt32(cfg.ReplayIndex),
		Format:          utils.PtrToString(cfg.Format),
		IgnoreErrs:      utils.PtrToBool(cfg.IgnoreErrs),
		PsClose:         utils.PtrToString(cfg.PsClose),
		BufSize:         utils.PtrToInt32(cfg.BufSize),
		Speed:           utils.PtrToString(cfg.Speed),
		DynamicInput:    utils.PtrToBool(cfg.DynamicInput),
		ReadOnly:        utils.PtrToBool(cfg.ReadOnly),
		CustomArguments: cfg.CustomArguments,
	}
}

func (c *replayInnerControllerImpl) entityReplayConfigs2PB(cfgs []entity.ReplayConfig) []*pbReplay.ReplayConfig {
	result := make([]*pbReplay.ReplayConfig, 0, len(cfgs))
	for _, cfg := range cfgs {
		result = append(result, c.entityReplayConfig2PB(cfg))
	}
	return result
}

func (c *replayInnerControllerImpl) entityTaskResources2PB(
	res *entity.ReplayTaskResources,
) *pbReplay.ReplayTaskResources {
	if res == nil {
		return nil
	}
	return &pbReplay.ReplayTaskResources{
		InstanceType: res.InstanceType,
		StorageSize:  res.StorageSize,
	}
}

func (c *replayInnerControllerImpl) entityReplayStatus2PB(status *entity.ReplayStatus) *pbReplay.ReplayStatus {
	if status == nil {
		return nil
	}

	pbStatus := &pbReplay.ReplayStatus{
		Phase:          c.entityPhase2PB(status.Phase),
		Message:        status.Message,
		AwsNodeRoleArn: status.AWSNodeRoleArn,
	}

	if status.LastUpdateTime != nil {
		pbStatus.LastUpdateTime = timestamppb.New(*status.LastUpdateTime)
	}

	if len(status.Tasks) > 0 {
		pbStatus.Tasks = c.entityTaskStatuses2PB(status.Tasks)
	}

	return pbStatus
}

func (c *replayInnerControllerImpl) entityTaskStatuses2PB(
	tasks []entity.ReplayTaskStatus,
) []*pbReplay.ReplayTaskStatus {
	result := make([]*pbReplay.ReplayTaskStatus, 0, len(tasks))
	for _, t := range tasks {
		result = append(result, &pbReplay.ReplayTaskStatus{
			Name:            t.Name,
			ReplicaStatuses: c.entityReplicaStatuses2PB(t.ReplicaStatuses),
			Message:         t.Message,
		})
	}
	return result
}

func (c *replayInnerControllerImpl) entityReplicaStatuses2PB(
	replicas []entity.ReplayReplicaStatus,
) []*pbReplay.ReplayReplicaStatus {
	result := make([]*pbReplay.ReplayReplicaStatus, 0, len(replicas))
	for _, r := range replicas {
		pbReplica := &pbReplay.ReplayReplicaStatus{
			PodName: r.PodName,
			Index:   r.Index,
			Phase:   c.entityTaskPhase2PB(r.Phase),
			Ready:   r.Ready,
			Message: r.Message,
		}
		if r.TaskInfo != nil {
			pbReplica.TaskInfo = c.entityTaskInfo2PB(r.TaskInfo)
		}
		result = append(result, pbReplica)
	}
	return result
}

func (c *replayInnerControllerImpl) entityTaskInfo2PB(info *entity.ReplayTaskInfo) *pbReplay.ReplayTaskInfo {
	if info == nil {
		return nil
	}
	return &pbReplay.ReplayTaskInfo{
		Type:                 info.Type,
		Status:               info.Status,
		StartTime:            info.StartTime,
		EndTime:              info.EndTime,
		Progress:             info.Progress,
		Error:                info.Error,
		LastCommandTimestamp: info.LastCommandTimestamp,
		SyncTime:             info.SyncTime,
		SourceAuditLogAddrs:  c.entityAuditLogAddrs2PB(info.SourceAuditLogAddrs),
		TotalCommands:        info.TotalCommands,
		ReplayedCommands:     info.ReplayedCommands,
		FailedCommands:       info.FailedCommands,
		TotalBytes:           info.TotalBytes,
		ProcessedBytes:       info.ProcessedBytes,
	}
}

func (c *replayInnerControllerImpl) entityDesiredState2PB(state entity.ReplayDesiredState) pbReplay.ReplayDesiredState {
	switch state {
	case entity.ReplayDesiredStateDeploy:
		return pbReplay.ReplayDesiredState_REPLAY_DESIRED_STATE_DEPLOY
	case entity.ReplayDesiredStateActive:
		return pbReplay.ReplayDesiredState_REPLAY_DESIRED_STATE_ACTIVE
	case entity.ReplayDesiredStatePause:
		return pbReplay.ReplayDesiredState_REPLAY_DESIRED_STATE_PAUSE
	default:
		return pbReplay.ReplayDesiredState_REPLAY_DESIRED_STATE_UNSPECIFIED
	}
}

func (c *replayInnerControllerImpl) entityCurrentState2PB(state entity.ReplayCurrentState) pbReplay.ReplayCurrentState {
	switch state {
	case entity.ReplayCurrentStatePending:
		return pbReplay.ReplayCurrentState_REPLAY_CURRENT_STATE_PENDING
	case entity.ReplayCurrentStateDeploying:
		return pbReplay.ReplayCurrentState_REPLAY_CURRENT_STATE_DEPLOYING
	case entity.ReplayCurrentStateDeployed:
		return pbReplay.ReplayCurrentState_REPLAY_CURRENT_STATE_DEPLOYED
	case entity.ReplayCurrentStateStarting:
		return pbReplay.ReplayCurrentState_REPLAY_CURRENT_STATE_STARTING
	case entity.ReplayCurrentStateRunning:
		return pbReplay.ReplayCurrentState_REPLAY_CURRENT_STATE_RUNNING
	case entity.ReplayCurrentStatePausing:
		return pbReplay.ReplayCurrentState_REPLAY_CURRENT_STATE_PAUSING
	case entity.ReplayCurrentStatePaused:
		return pbReplay.ReplayCurrentState_REPLAY_CURRENT_STATE_PAUSED
	case entity.ReplayCurrentStateDeleting:
		return pbReplay.ReplayCurrentState_REPLAY_CURRENT_STATE_DELETING
	case entity.ReplayCurrentStateDeleted:
		return pbReplay.ReplayCurrentState_REPLAY_CURRENT_STATE_DELETED
	default:
		return pbReplay.ReplayCurrentState_REPLAY_CURRENT_STATE_UNSPECIFIED
	}
}

func (c *replayInnerControllerImpl) entityPhase2PB(phase entity.ReplayPhase) pbReplay.ReplayPhase {
	switch phase {
	case entity.ReplayPhasePending:
		return pbReplay.ReplayPhase_REPLAY_PHASE_PENDING
	case entity.ReplayPhaseCreating:
		return pbReplay.ReplayPhase_REPLAY_PHASE_CREATING
	case entity.ReplayPhaseRunning:
		return pbReplay.ReplayPhase_REPLAY_PHASE_RUNNING
	case entity.ReplayPhaseStopped:
		return pbReplay.ReplayPhase_REPLAY_PHASE_STOPPED
	case entity.ReplayPhaseDeleting:
		return pbReplay.ReplayPhase_REPLAY_PHASE_DELETED
	default:
		return pbReplay.ReplayPhase_REPLAY_PHASE_UNSPECIFIED
	}
}

func (c *replayInnerControllerImpl) entityTaskPhase2PB(phase entity.ReplayTaskPhase) pbReplay.ReplayTaskPhase {
	switch phase {
	case entity.ReplayTaskPhasePending:
		return pbReplay.ReplayTaskPhase_REPLAY_TASK_PHASE_PENDING
	case entity.ReplayTaskPhaseRunning:
		return pbReplay.ReplayTaskPhase_REPLAY_TASK_PHASE_RUNNING
	case entity.ReplayTaskPhaseStopped:
		return pbReplay.ReplayTaskPhase_REPLAY_TASK_PHASE_STOPPED
	case entity.ReplayTaskPhaseDeleted:
		return pbReplay.ReplayTaskPhase_REPLAY_TASK_PHASE_DELETED
	default:
		return pbReplay.ReplayTaskPhase_REPLAY_TASK_PHASE_UNSPECIFIED
	}
}

func (c *replayInnerControllerImpl) validateCreateReplayRequest(req *pbReplay.CreateReplayRequest) bizerr.BizError {
	if req.OrgId == "" {
		return bizerr.New(bizerr.ErrCodeParams, "org_id is required")
	}
	if req.ClusterId == "" {
		return bizerr.New(bizerr.ErrCodeParams, "cluster_id is required")
	}
	if req.Replay == nil {
		return bizerr.New(bizerr.ErrCodeParams, "replay spec is required")
	}

	spec := req.Replay
	if spec.SourceClusterId == "" {
		return bizerr.New(bizerr.ErrCodeParams, "source_cluster_id is required")
	}
	if spec.TargetClusterId == "" {
		return bizerr.New(bizerr.ErrCodeParams, "target_cluster_id is required")
	}
	// Validate replay tasks if provided
	if len(spec.ReplayTasks) > 0 {
		for i, task := range spec.ReplayTasks {
			if task.Name == "" {
				return bizerr.New(bizerr.ErrCodeParams, "replay_tasks[%d].name is required", i)
			}
			if task.DataSource == nil {
				return bizerr.New(bizerr.ErrCodeParams, "replay_tasks[%d].data_source is required", i)
			}
			if task.DataTarget == nil {
				return bizerr.New(bizerr.ErrCodeParams, "replay_tasks[%d].data_target is required", i)
			}
			if task.DataTarget.Endpoint == "" {
				return bizerr.New(bizerr.ErrCodeParams, "replay_tasks[%d].data_target.endpoint is required", i)
			}
		}
	}

	return nil
}

func (c *replayInnerControllerImpl) validateUpdateReplayRequest(req *pbReplay.UpdateReplayRequest) bizerr.BizError {
	if req.OrgId == "" {
		return bizerr.New(bizerr.ErrCodeParams, "org_id is required")
	}
	if req.ClusterId == "" {
		return bizerr.New(bizerr.ErrCodeParams, "cluster_id is required")
	}
	if req.ReplayId == "" {
		return bizerr.New(bizerr.ErrCodeParams, "replay_id is required")
	}
	if req.Replay == nil {
		return bizerr.New(bizerr.ErrCodeParams, "replay spec is required")
	}

	// Validate replay tasks if provided
	spec := req.Replay
	if len(spec.ReplayTasks) > 0 {
		for i, task := range spec.ReplayTasks {
			if task.Name == "" {
				return bizerr.New(bizerr.ErrCodeParams, "replay_tasks[%d].name is required", i)
			}
			if task.DataTarget != nil && task.DataTarget.Endpoint == "" {
				return bizerr.New(bizerr.ErrCodeParams, "replay_tasks[%d].data_target.endpoint is required", i)
			}
		}
	}

	return nil
}