package cluster

import (
	"encoding/json"
	"fmt"
	"time"

	"github.com/gin-gonic/gin"
	rpproto "github.com/tidbcloud/dataflow-service/idl/pbgen/proto/replay"
	"google.golang.org/protobuf/encoding/protojson"
	"google.golang.org/protobuf/types/known/timestamppb"

	"github.com/tidbcloud/devops/api"
	clusterapi "github.com/tidbcloud/devops/api/cluster"
	opent "github.com/tidbcloud/devops/entity/operation"
	ctrl "github.com/tidbcloud/devops/pkg/controller"
	"github.com/tidbcloud/devops/pkg/infra/consts"
	"github.com/tidbcloud/devops/pkg/infra/utils/container"
	opsvcapi "github.com/tidbcloud/devops/pkg/service/operation"
	opsvcclusteraction "github.com/tidbcloud/devops/pkg/service/operation/actions/clusteraction"
	opsvcins "github.com/tidbcloud/devops/pkg/service/operation/instance"
)

// ListTrafficReplays
//
//	@Id				ClusterServiceListTrafficReplays
//	@Description	list replays
//	@Tags			ClusterService
//	@Accept			json
//	@Produce		json
//	@Param			req	query		clusterapi.ListTrafficReplaysRequest	true	"ListTasksReq"
//	@Success		200	{object}	clusterapi.ListTrafficReplaysResponse
//	@Failure		500	{object}	api.BaseResp
//	@Router			/api/v1/devops/cluster/traffic/replays [get]
func (c *ClusterController) ListTrafficReplays(ctx *gin.Context) {
	req := &clusterapi.ListTrafficReplaysRequest{}
	if err := ctx.ShouldBindQuery(req); err != nil {
		ctrl.FailWithCode(ctx, err, api.ErrCodeInvalidArgument)
		return
	}

	reqPB := &rpproto.ListReplaysRequest{
		OrgId:     req.TenantID,
		ClusterId: req.ClusterID,
		Page:      int32(req.Page),
		PageSize:  int32(req.PerPage),
	}
	resp, err := c.dataflowCli.Replay.ListReplays(ctx, reqPB)
	if err != nil {
		ctrl.Fail(ctx, err)
		return
	}

	replays := container.TransformSlice(resp.GetReplays(), func(rpProto *rpproto.Replay) *clusterapi.TrafficReplay {
		return TrafficReplayConverter{}.ReplayFromPB(rpProto)
	})
	ctrl.Success(ctx, &clusterapi.ListTrafficReplaysResponse{
		Replays: replays,
		Total:   int64(resp.TotalSize),
	})
}

// GetTrafficReplay
//
//	@Id				ClusterServiceGetTrafficReplay
//	@Description	get a replay
//	@Tags			ClusterService
//	@Accept			json
//	@Produce		json
//	@Param			replay_id	path		string								true	"replay id"
//	@Param			req			query		clusterapi.GetTrafficReplayRequest	true	"GetReplayRequest"
//	@Success		200			{object}	clusterapi.GetTrafficReplayResponse
//	@Failure		500			{object}	api.BaseResp
//	@Router			/api/v1/devops/cluster/traffic/replays/{replay_id} [get]
func (c *ClusterController) GetTrafficReplay(ctx *gin.Context) {
	replayID := ctx.Param("replay_id")
	req := &clusterapi.GetTrafficReplayRequest{}
	if err := ctx.ShouldBindQuery(req); err != nil {
		ctrl.FailWithCode(ctx, err, api.ErrCodeInvalidArgument)
		return
	}

	reqPB := &rpproto.GetReplayRequest{
		OrgId:     req.TenantID,
		ClusterId: req.ClusterID,
		ReplayId:  replayID,
	}
	resp, err := c.dataflowCli.Replay.GetReplay(ctx, reqPB)
	if err != nil {
		ctrl.Fail(ctx, err)
		return
	}

	ctrl.Success(ctx, &clusterapi.GetTrafficReplayResponse{
		Replay: TrafficReplayConverter{}.ReplayFromPB(resp.GetReplay()),
	})
}

// CreateTrafficReplay
//
//	@Id				ClusterServiceCreateTrafficReplay
//	@Description	create a replay
//	@Tags			ClusterService
//	@Accept			json
//	@Produce		json
//	@Param			req	body		clusterapi.CreateTrafficReplayRequest	true	"CreateReplayRequest"
//	@Success		200	{object}	clusterapi.CreateTrafficReplayResponse
//	@Failure		500	{object}	api.BaseResp
//	@Router			/api/v1/devops/cluster/traffic/replays [post]
func (c *ClusterController) CreateTrafficReplay(ctx *gin.Context) {
	req := &clusterapi.CreateTrafficReplayRequest{}
	if err := ctx.ShouldBindJSON(req); err != nil {
		ctrl.FailWithCode(ctx, err, api.ErrCodeInvalidArgument)
		return
	}

	userEmail := ctx.GetString(consts.CtxUserEmail)
	if userEmail == "" {
		ctrl.FailWithCode(ctx, fmt.Errorf("account not found in ctx"), api.ErrCodeInvalidArgument)
		return
	}

	specPB, err := TrafficReplayConverter{}.ReplaySpecToPB(req.ReplaySpec)
	if err != nil {
		ctrl.Fail(ctx, err)
		return
	}
	reqPB := &rpproto.CreateReplayRequest{
		OrgId:     req.TenantID,
		ClusterId: req.ClusterID,
		Replay:    specPB,
	}

	opReqData, err := protojson.Marshal(reqPB)
	if err != nil {
		ctrl.FailWithCode(ctx, fmt.Errorf("marshal proto message failed: %s", err), api.ErrCodeInternal)
		return
	}
	createOPParam := opsvcapi.CreateInstanceParam{
		Type:      opent.OperationTargetCluster,
		Operation: "CreateClusterTrafficReplay", // NOTE: MUST match the name in operation definition.
		Reason:    "Create traffic replay",
		Applicant: userEmail,
		Params: map[opent.ParamName]string{
			opent.ParamClusterID:                  req.ClusterID,
			opent.ParamCreateTrafficReplayRequest: string(opReqData),
		},
		Actions: []opsvcins.CreateInstanceActionParam{
			{
				Name:        opsvcclusteraction.ActionCreateTrafficReplay.Name,
				StartPolicy: opent.ActionStartPolicyManual,
			},
		},
	}

	insID, err := c.operationSvc.CreateInstance(ctx, createOPParam)
	if err != nil {
		ctrl.Fail(ctx, fmt.Errorf("create operation instance failed: %w", err))
		return
	}

	resp := &clusterapi.CreateTrafficReplayResponse{
		InstanceID: *insID,
	}
	ctrl.Success(ctx, resp)
}

// UpdateTrafficReplay
//
//	@Id				ClusterServiceUpdateTrafficReplay
//	@Description	update a replay
//	@Tags			ClusterService
//	@Accept			json
//	@Produce		json
//	@Param			replay_id	path		string									true	"replay id"
//	@Param			req			body		clusterapi.UpdateTrafficReplayRequest	true	"UpdateReplayRequest"
//	@Success		200			{object}	clusterapi.UpdateTrafficReplayResponse
//	@Failure		500			{object}	api.BaseResp
//	@Router			/api/v1/devops/cluster/traffic/replays/{replay_id} [put]
func (c *ClusterController) UpdateTrafficReplay(ctx *gin.Context) {
	replayID := ctx.Param("replay_id")
	req := &clusterapi.UpdateTrafficReplayRequest{}
	if err := ctx.ShouldBindJSON(req); err != nil {
		ctrl.FailWithCode(ctx, err, api.ErrCodeInvalidArgument)
		return
	}

	userEmail := ctx.GetString(consts.CtxUserEmail)
	if userEmail == "" {
		ctrl.FailWithCode(ctx, fmt.Errorf("account not found in ctx"), api.ErrCodeInvalidArgument)
		return
	}

	specPB, err := TrafficReplayConverter{}.ReplaySpecToPB(req.ReplaySpec)
	if err != nil {
		ctrl.Fail(ctx, fmt.Errorf("convert spec failed: %s", err))
		return
	}

	reqPB := &rpproto.UpdateReplayRequest{
		OrgId:     req.TenantID,
		ClusterId: req.ClusterID,
		ReplayId:  replayID,
		Replay:    specPB,
	}
	opReqData, err := protojson.Marshal(reqPB)
	if err != nil {
		ctrl.FailWithCode(ctx, fmt.Errorf("marshal proto message failed: %s", err), api.ErrCodeInternal)
		return
	}

	createOPParam := opsvcapi.CreateInstanceParam{
		Type:      opent.OperationTargetCluster,
		Operation: "UpdateClusterTrafficReplay", // NOTE: MUST match the name in operation definition.
		Reason:    "Update traffic replay",
		Applicant: userEmail,
		Params: map[opent.ParamName]string{
			opent.ParamClusterID:                         req.ClusterID,
			opent.ParamTrafficReplayID:                   replayID,
			opent.ParamUpdateTrafficReplayRequest:        string(opReqData),
			opent.ParamUpdateTrafficReplayRequestChanged: req.ChangedReplaySpec,
		},
		Actions: []opsvcins.CreateInstanceActionParam{
			{
				Name:        opsvcclusteraction.ActionUpdateTrafficReplay.Name,
				StartPolicy: opent.ActionStartPolicyManual,
			},
		},
	}

	insID, err := c.operationSvc.CreateInstance(ctx, createOPParam)
	if err != nil {
		ctrl.Fail(ctx, fmt.Errorf("create operation instance failed: %w", err))
		return
	}

	resp := &clusterapi.UpdateTrafficReplayResponse{
		InstanceID: *insID,
	}
	ctrl.Success(ctx, resp)
}

// DeleteTrafficReplay
//
//	@Id				ClusterServiceDeleteTrafficReplay
//	@Description	delete a replay
//	@Tags			ClusterService
//	@Accept			json
//	@Produce		json
//	@Param			replay_id	path		string									true	"replay id"
//	@Param			req			query		clusterapi.DeleteTrafficReplayRequest	true	"DeleteReplayRequest"
//	@Success		200			{object}	clusterapi.DeleteTrafficReplayResponse
//	@Failure		500			{object}	api.BaseResp
//	@Router			/api/v1/devops/cluster/traffic/replays/{replay_id} [delete]
func (c *ClusterController) DeleteTrafficReplay(ctx *gin.Context) {
	replayID := ctx.Param("replay_id")
	req := &clusterapi.DeleteTrafficReplayRequest{}
	if err := ctx.ShouldBindQuery(req); err != nil {
		ctrl.FailWithCode(ctx, err, api.ErrCodeInvalidArgument)
		return
	}

	userEmail := ctx.GetString(consts.CtxUserEmail)
	if userEmail == "" {
		ctrl.FailWithCode(ctx, fmt.Errorf("account not found in ctx"), api.ErrCodeInvalidArgument)
		return
	}

	reqPB := &rpproto.DeleteReplayRequest{
		OrgId:     req.TenantID,
		ClusterId: req.ClusterID,
		ReplayId:  replayID,
	}
	opReqData, err := protojson.Marshal(reqPB)
	if err != nil {
		ctrl.FailWithCode(ctx, fmt.Errorf("marshal proto message failed: %s", err), api.ErrCodeInternal)
		return
	}

	createOPParam := opsvcapi.CreateInstanceParam{
		Type:      opent.OperationTargetCluster,
		Operation: "DeleteClusterTrafficReplay", // NOTE: MUST match the name in operation definition.
		Reason:    "Delete traffic replay",
		Applicant: userEmail,
		Params: map[opent.ParamName]string{
			opent.ParamClusterID:                  req.ClusterID,
			opent.ParamTrafficReplayID:            replayID,
			opent.ParamDeleteTrafficReplayRequest: string(opReqData),
		},
		Actions: []opsvcins.CreateInstanceActionParam{
			{
				Name:        opsvcclusteraction.ActionDeleteTrafficReplay.Name,
				StartPolicy: opent.ActionStartPolicyManual,
			},
		},
	}

	insID, err := c.operationSvc.CreateInstance(ctx, createOPParam)
	if err != nil {
		ctrl.Fail(ctx, fmt.Errorf("create operation instance failed: %w", err))
		return
	}

	resp := &clusterapi.DeleteTrafficReplayResponse{
		InstanceID: *insID,
	}
	ctrl.Success(ctx, resp)
}

// StartTrafficReplay
//
//	@Id				ClusterServiceStartTrafficReplay
//	@Description	start a replay
//	@Tags			ClusterService
//	@Accept			json
//	@Produce		json
//	@Param			replay_id	path		string									true	"replay id"
//	@Param			req			body		clusterapi.StartTrafficReplayRequest	true	"StartReplayRequest"
//	@Success		200			{object}	clusterapi.StartTrafficReplayResponse
//	@Failure		500			{object}	api.BaseResp
//	@Router			/api/v1/devops/cluster/traffic/replays/{replay_id}/start [post]
func (c *ClusterController) StartTrafficReplay(ctx *gin.Context) {
	replayID := ctx.Param("replay_id")
	req := &clusterapi.StartTrafficReplayRequest{}
	if err := ctx.ShouldBindJSON(req); err != nil {
		ctrl.FailWithCode(ctx, err, api.ErrCodeInvalidArgument)
		return
	}

	userEmail := ctx.GetString(consts.CtxUserEmail)
	if userEmail == "" {
		ctrl.FailWithCode(ctx, fmt.Errorf("account not found in ctx"), api.ErrCodeInvalidArgument)
		return
	}

	reqPB := &rpproto.StartReplayRequest{
		OrgId:     req.TenantID,
		ClusterId: req.ClusterID,
		ReplayId:  replayID,
	}
	opReqData, err := protojson.Marshal(reqPB)
	if err != nil {
		ctrl.FailWithCode(ctx, fmt.Errorf("marshal proto message failed: %s", err), api.ErrCodeInternal)
		return
	}

	createOPParam := opsvcapi.CreateInstanceParam{
		Type:      opent.OperationTargetCluster,
		Operation: "StartClusterTrafficReplay", // NOTE: MUST match the name in operation definition.
		Reason:    "Start traffic replay",
		Applicant: userEmail,
		Params: map[opent.ParamName]string{
			opent.ParamClusterID:                 req.ClusterID,
			opent.ParamTrafficReplayID:           replayID,
			opent.ParamStartTrafficReplayRequest: string(opReqData),
		},
		Actions: []opsvcins.CreateInstanceActionParam{
			{
				Name:        opsvcclusteraction.ActionStartTrafficReplay.Name,
				StartPolicy: opent.ActionStartPolicyManual,
			},
		},
	}

	insID, err := c.operationSvc.CreateInstance(ctx, createOPParam)
	if err != nil {
		ctrl.Fail(ctx, fmt.Errorf("create operation instance failed: %w", err))
		return
	}

	resp := &clusterapi.StartTrafficReplayResponse{
		InstanceID: *insID,
	}
	ctrl.Success(ctx, resp)
}

// StopTrafficReplay
//
//	@Id				ClusterServiceStopTrafficReplay
//	@Description	stop a replay
//	@Tags			ClusterService
//	@Accept			json
//	@Produce		json
//	@Param			replay_id	path		string								true	"replay id"
//	@Param			req			body		clusterapi.StopTrafficReplayRequest	true	"StopReplayRequest"
//	@Success		200			{object}	clusterapi.StopTrafficReplayResponse
//	@Failure		500			{object}	api.BaseResp
//	@Router			/api/v1/devops/cluster/traffic/replays/{replay_id}/stop [post]
func (c *ClusterController) StopTrafficReplay(ctx *gin.Context) {
	replayID := ctx.Param("replay_id")
	req := &clusterapi.StopTrafficReplayRequest{}
	if err := ctx.ShouldBindJSON(req); err != nil {
		ctrl.FailWithCode(ctx, err, api.ErrCodeInvalidArgument)
		return
	}

	userEmail := ctx.GetString(consts.CtxUserEmail)
	if userEmail == "" {
		ctrl.FailWithCode(ctx, fmt.Errorf("account not found in ctx"), api.ErrCodeInvalidArgument)
		return
	}

	reqPB := &rpproto.StopReplayRequest{
		OrgId:     req.TenantID,
		ClusterId: req.ClusterID,
		ReplayId:  replayID,
	}
	opReqData, err := protojson.Marshal(reqPB)
	if err != nil {
		ctrl.FailWithCode(ctx, fmt.Errorf("marshal proto message failed: %s", err), api.ErrCodeInternal)
		return
	}

	createOPParam := opsvcapi.CreateInstanceParam{
		Type:      opent.OperationTargetCluster,
		Operation: "StopClusterTrafficReplay", // NOTE: MUST match the name in operation definition.
		Reason:    "Stop traffic replay",
		Applicant: userEmail,
		Params: map[opent.ParamName]string{
			opent.ParamClusterID:                req.ClusterID,
			opent.ParamTrafficReplayID:          replayID,
			opent.ParamStopTrafficReplayRequest: string(opReqData),
		},
		Actions: []opsvcins.CreateInstanceActionParam{
			{
				Name:        opsvcclusteraction.ActionStopTrafficReplay.Name,
				StartPolicy: opent.ActionStartPolicyManual,
			},
		},
	}

	insID, err := c.operationSvc.CreateInstance(ctx, createOPParam)
	if err != nil {
		ctrl.Fail(ctx, fmt.Errorf("create operation instance failed: %w", err))
		return
	}

	resp := &clusterapi.StopTrafficReplayResponse{
		InstanceID: *insID,
	}
	ctrl.Success(ctx, resp)
}

type TrafficReplayConverter struct{}

func (c TrafficReplayConverter) ReplayFromPB(pb *rpproto.Replay) *clusterapi.TrafficReplay {
	if pb == nil {
		return nil
	}

	replay := &clusterapi.TrafficReplay{
		ReplayID:              pb.GetReplayId(),
		OrgID:                 pb.GetOrgId(),
		ProjectID:             pb.GetProjectId(),
		ClusterID:             pb.GetClusterId(),
		DesiredState:          pb.GetDesiredState().String(),
		CurrentState:          pb.GetCurrentState().String(),
		SourceClusterID:       pb.GetSourceClusterId(),
		CloudProvider:         pb.GetCloudProvider(),
		Region:                pb.GetRegion(),
		ReplayTasks:           c.ReplayTasksFromPB(pb.GetReplayTasks()),
		ReplayConfig:          nil,
		Replicas:              pb.GetReplicas(),
		TaskResources:         c.TaskResourcesFromPB(pb.GetTaskResources()),
		ReplayImage:           pb.GetReplayImage(),
		SourceAuditCredential: c.AuditLogAddrFromPB(pb.GetSourceAuditCredential()),
		ReplayStatus:          c.ReplayStatusFromPB(pb.GetReplayStatus()),
		CreatedAt:             timestampAsTime(pb.GetCreatedAt()),
		UpdatedAt:             timestampAsTime(pb.GetUpdatedAt()),
	}

	if pb.ReplayConfig != nil {
		replay.ReplayConfig = &clusterapi.TrafficReplayCommonConfig{
			Username: pb.ReplayConfig.GetUsername(),
			Password: pb.ReplayConfig.GetPassword(),
		}
		if pb.ReplayConfig.ReplayConfig != nil {
			replay.ReplayConfig.ReplayConfig = c.ReplayConfigFromPB(pb.ReplayConfig.ReplayConfig)
		}
	}
	return replay
}

func timestampAsTime(ts *timestamppb.Timestamp) time.Time {
	return ts.AsTime()
}

func (c TrafficReplayConverter) ReplayTasksFromPB(pbs []*rpproto.ReplayTask) []*clusterapi.TrafficReplayTask {
	if len(pbs) == 0 {
		return nil
	}
	tasks := make([]*clusterapi.TrafficReplayTask, len(pbs))
	for i, pb := range pbs {
		tasks[i] = &clusterapi.TrafficReplayTask{
			Name: pb.GetName(),
			DataSource: &clusterapi.TrafficReplayDataSource{
				SourceAuditLogAddrs: container.TransformSlice(pb.GetDataSource().GetSourceAuditLogAddrs(), func(adpb *rpproto.AuditLogAddr) *clusterapi.TrafficReplayAuditLogAddr {
					return c.AuditLogAddrFromPB(adpb)
				}),
			},
			DataTarget: &clusterapi.TrafficReplayDataTarget{
				Endpoint: pb.GetDataTarget().GetEndpoint(),
				Username: pb.GetDataTarget().GetUsername(),
				Password: pb.GetDataTarget().GetPassword(),
			},
			Replicas:        pb.GetReplicas(),
			ReplayStartTime: pb.GetReplayStartTime(),
			RunStartTime:    pb.GetRunStartTime(),
			Resources:       c.TaskResourcesFromPB(pb.GetResources()),
			PerReplayConfigs: container.TransformSlice(pb.GetPerReplayConfigs(), func(cfgpb *rpproto.ReplayConfig) *clusterapi.TrafficReplayConfig {
				return c.ReplayConfigFromPB(cfgpb)
			}),
			Image: pb.GetImage(),
		}
	}
	return tasks
}

func (c TrafficReplayConverter) ReplayConfigFromPB(pb *rpproto.ReplayConfig) *clusterapi.TrafficReplayConfig {
	if pb == nil {
		return nil
	}
	return &clusterapi.TrafficReplayConfig{
		ReplayStartTime: pb.GetReplayStartTime(),
		RunStartTime:    pb.GetRunStartTime(),
		ReplayIndex:     pb.GetReplayIndex(),
		CustomArguments: pb.GetCustomArguments(),
		Format:          pb.GetFormat(),
		IgnoreErrs:      pb.GetIgnoreErrs(),
		PsClose:         pb.GetPsClose(),
		BufSize:         pb.GetBufSize(),
		Speed:           pb.GetSpeed(),
		DynamicInput:    pb.GetDynamicInput(),
		ReadOnly:        pb.GetReadOnly(),
	}
}

func (c TrafficReplayConverter) TaskResourcesFromPB(pb *rpproto.ReplayTaskResources) *clusterapi.TrafficReplayTaskResources {
	if pb == nil {
		return nil
	}
	return &clusterapi.TrafficReplayTaskResources{
		InstanceType: pb.GetInstanceType(),
		StorageSize:  pb.GetStorageSize(),
	}
}

func (c TrafficReplayConverter) AuditLogAddrFromPB(pb *rpproto.AuditLogAddr) *clusterapi.TrafficReplayAuditLogAddr {
	if pb == nil {
		return nil
	}

	addr := &clusterapi.TrafficReplayAuditLogAddr{}
	switch x := pb.GetStorage().(type) {
	case *rpproto.AuditLogAddr_Aws:
		addr.AWS = &clusterapi.TrafficReplayAWSAuditLog{
			S3Url:      x.Aws.GetS3Url(),
			RoleArn:    x.Aws.GetRoleArn(),
			ExternalID: x.Aws.GetExternalId(),
		}
	case *rpproto.AuditLogAddr_Gcp:
		addr.GCP = &clusterapi.TrafficReplayGCPAuditLog{
			GcsUrl:             x.Gcp.GetGcsUrl(),
			ServiceAccountJson: x.Gcp.GetServiceAccountJson(),
		}
	case *rpproto.AuditLogAddr_Aliyun:
		addr.Aliyun = &clusterapi.TrafficReplayAliyunAuditLog{
			BucketURL:       x.Aliyun.GetBucketUrl(),
			AccessKeyID:     x.Aliyun.GetAccessKeyId(),
			SecretAccessKey: x.Aliyun.GetSecretAccessKey(),
		}
	}
	return addr
}

func (c TrafficReplayConverter) AuditLogAddrToPB(addr *clusterapi.TrafficReplayAuditLogAddr) (*rpproto.AuditLogAddr, error) {
	if addr == nil {
		return nil, nil
	}

	pb := &rpproto.AuditLogAddr{}
	if addr.AWS != nil {
		pb.Storage = &rpproto.AuditLogAddr_Aws{
			Aws: &rpproto.AWSAuditLog{
				S3Url:      addr.AWS.S3Url,
				RoleArn:    addr.AWS.RoleArn,
				ExternalId: addr.AWS.ExternalID,
			},
		}
	}
	if addr.GCP != nil {
		pb.Storage = &rpproto.AuditLogAddr_Gcp{
			Gcp: &rpproto.GCPAuditLog{
				GcsUrl:             addr.GCP.GcsUrl,
				ServiceAccountJson: addr.GCP.ServiceAccountJson,
			},
		}
	}
	if addr.Aliyun != nil {
		pb.Storage = &rpproto.AuditLogAddr_Aliyun{
			Aliyun: &rpproto.AliyunAuditLog{
				BucketUrl:       addr.Aliyun.BucketURL,
				AccessKeyId:     addr.Aliyun.AccessKeyID,
				SecretAccessKey: addr.Aliyun.SecretAccessKey,
			},
		}
	}
	return pb, nil
}

func (c TrafficReplayConverter) ReplayTaskInfoFromPB(pb *rpproto.ReplayTaskInfo) *clusterapi.TrafficReplayTaskInfo {
	if pb == nil {
		return nil
	}
	return &clusterapi.TrafficReplayTaskInfo{
		Type:                 pb.GetType(),
		Status:               pb.GetStatus(),
		StartTime:            pb.GetStartTime(),
		EndTime:              pb.GetEndTime(),
		Progress:             pb.GetProgress(),
		Error:                pb.GetError(),
		LastCommandTimestamp: pb.GetLastCommandTimestamp(),
		SyncTime:             pb.GetSyncTime(),
		SourceAuditLogAddrs: container.TransformSlice(pb.GetSourceAuditLogAddrs(), func(adpb *rpproto.AuditLogAddr) *clusterapi.TrafficReplayAuditLogAddr {
			return c.AuditLogAddrFromPB(adpb)
		}),
		TotalCommands: pb.GetTotalCommands(),
	}
}

func (c TrafficReplayConverter) ReplaySpecToPB(spec *clusterapi.TrafficReplaySpec) (*rpproto.ReplaySpec, error) {
	if spec == nil {
		return nil, fmt.Errorf("replay spec is required")
	}

	specPB := &rpproto.ReplaySpec{
		SourceClusterId: spec.SourceClusterID,
		TargetClusterId: spec.TargetClusterID,
		ProjectId:       spec.ProjectID,
		Replicas:        spec.Replicas,
		ReplayImage:     spec.ReplayImage,
		CloudProvider:   spec.CloudProvider,
		Region:          spec.Region,
		ReplayConfig: &rpproto.ReplayCommonConfig{
			Username: spec.DatabaseUsername,
			Password: spec.DatabasePassword,
			ReplayConfig: &rpproto.ReplayConfig{
				ReplayStartTime: spec.ConfigReplayStartTime,
				RunStartTime:    spec.ConfigRunStartTime,
				CustomArguments: spec.ConfigCustomArguments,
			},
		},
	}

	if spec.TaskResource != nil {
		specPB.TaskResources = &rpproto.ReplayTaskResources{
			InstanceType: spec.TaskResource.InstanceType,
			StorageSize:  spec.TaskResource.StorageSize,
		}
	}

	if spec.SourceAuditCredential != nil {
		sourceAuditCredential, err := c.AuditLogAddrToPB(spec.SourceAuditCredential)
		if err != nil {
			return nil, fmt.Errorf("source_audit_credential is invalid: %w", err)
		}
		specPB.SourceAuditCredential = sourceAuditCredential
	}

	if len(spec.ReplayTasks) > 0 {
		var rawTasks []json.RawMessage
		if err := json.Unmarshal([]byte(spec.ReplayTasks), &rawTasks); err != nil {
			return nil, fmt.Errorf("unmarshal replay task string failed: %s", err)
		}

		tasksPB := make([]*rpproto.ReplayTask, 0, len(rawTasks))
		for _, raw := range rawTasks {
			task := &rpproto.ReplayTask{}
			if err := protojson.Unmarshal(raw, task); err != nil {
				return nil, fmt.Errorf("unmarshal replay task failed: %s", err)
			}
			tasksPB = append(tasksPB, task)
		}
		specPB.ReplayTasks = tasksPB
	}

	return specPB, nil
}

func (c TrafficReplayConverter) ReplayStatusFromPB(pb *rpproto.ReplayStatus) *clusterapi.TrafficReplayStatus {
	return &clusterapi.TrafficReplayStatus{
		Phase:   pb.GetPhase().String(),
		Message: pb.GetMessage(),
		Tasks: container.TransformSlice(pb.GetTasks(), func(pbss *rpproto.ReplayTaskStatus) *clusterapi.TrafficReplayTaskStatus {
			return &clusterapi.TrafficReplayTaskStatus{
				Name: pbss.GetName(),
				ReplicaStatuses: container.TransformSlice(pbss.GetReplicaStatuses(), func(rrs *rpproto.ReplayReplicaStatus) *clusterapi.TrafficReplayReplicaStatus {
					return &clusterapi.TrafficReplayReplicaStatus{
						PodName:  rrs.GetPodName(),
						Index:    rrs.GetIndex(),
						Phase:    rrs.GetPhase().String(),
						Ready:    rrs.GetReady(),
						Message:  rrs.GetMessage(),
						TaskInfo: c.ReplayTaskInfoFromPB(rrs.GetTaskInfo()),
					}
				}),
				Message: pbss.GetMessage(),
			}
		}),
		LastUpdateTime: timestampAsTime(pb.GetLastUpdateTime()),
		AwsNodeRoleArn: pb.GetAwsNodeRoleArn(),
	}
}