package dm

import (
	"context"
	"fmt"
	"strings"
	"sync"
	"time"

	"github.com/pingcap/errcode"
	"github.com/pingcap/errors"
	"github.com/pingcap/tiflow/dm/openapi"
	"github.com/pingcap/tiflow/dm/pkg/checker"
	"github.com/tidbcloud/core-svc/idl/pbgen/proto/core"
	entity2 "github.com/tidbcloud/dataflow-service/app/repo/entity"
	"github.com/tidbcloud/dataflow-service/idl/pbgen/proto/agent"
	"github.com/tidbcloud/dataflow-service/idl/pbgen/proto/dataflow"
	dmpb "github.com/tidbcloud/dataflow-service/idl/pbgen/proto/dm"
	"github.com/tidbcloud/dataflow-service/internal/bizutils"
	"github.com/tidbcloud/dataflow-service/internal/common"
	commondm "github.com/tidbcloud/dataflow-service/internal/common/dm"
	"github.com/tidbcloud/dataflow-service/internal/db/entity"
	"github.com/tidbcloud/dataflow-service/internal/grpc/grpcli"
	innerAgent "github.com/tidbcloud/dataflow-service/internal/handler/agent"
	"github.com/tidbcloud/dataflow-service/internal/model/connprofile"
	"github.com/tidbcloud/dataflow-service/internal/model/dm"
	"github.com/tidbcloud/dataflow-service/internal/model/feature"
	"github.com/tidbcloud/dataflow-service/internal/service"
	"github.com/tidbcloud/dataflow-service/internal/utils"
	"github.com/tidbcloud/pkgv2/log"
	"github.com/tidbcloud/pkgv2/logid"
	"github.com/tidbcloud/tiinf/sdk/auditlog"
	"go.uber.org/zap"
	"google.golang.org/protobuf/types/known/emptypb"
	"google.golang.org/protobuf/types/known/timestamppb"
	"gorm.io/gorm"
)

type dmService struct {
	dmpb.UnimplementedServiceServer
	service.OperationsService
	model dm.Model

	connProfileService dataflow.ConnProfileServiceServer
}

func NewService(opSvc service.OperationsService, model dm.Model, connProfileService dataflow.ConnProfileServiceServer) dmpb.ServiceServer {
	return &dmService{
		OperationsService:  opSvc,
		model:              model,
		connProfileService: connProfileService,
	}
}

func (s *dmService) userCtx(ctx context.Context) dm.UserContext {
	user := s.CurrentUser(ctx)
	return s.model.UserContext(ctx, user)
}

func (s *dmService) GetSupportedSources(ctx context.Context, req *dmpb.GetSupportedSourcesReq) (*dmpb.GetSupportedSourcesResp, error) {
	return &dmpb.GetSupportedSourcesResp{
		Sources: commondm.SupportedSources(req.GetVendor()),
	}, nil
}

func (s *dmService) ListJobSpecs(ctx context.Context, _ *emptypb.Empty) (*dmpb.ListJobSpecsResp, error) {
	userCtx := s.userCtx(ctx)
	logger := userCtx.Log()
	jobSpecs, err := dm.ListJobSpecs()
	if err != nil {
		logger.Error("failed to get job specs", zap.Error(err))
		return nil, errcode.NewInternalErr(errors.Errorf("Failed to get job specs"))
	}

	resp := &dmpb.ListJobSpecsResp{}
	for _, spec := range jobSpecs {
		resp.Specs = append(resp.Specs, jobSpecEntity2PB(&spec))
	}
	return resp, nil
}

func (s *dmService) checkMaintenance(dfJob *entity.DFJob, logger *zap.Logger, op string) error {
	if feature.IsFeatureEnabled(feature.MaintainingDMJobs, dfJob.TenantID.Uint64(), dfJob.ProjectID.Uint64(), dfJob.ClusterID.Uint64(), dfJob.Region, dfJob.Provider) {
		logger.Info("dm is maintaining, cannot do this", zap.Uint64("project_id", dfJob.ProjectID.Uint64()),
			zap.Uint64("cluster_id", dfJob.ClusterID.Uint64()), zap.String("operation", op))
		err := errors.New("DM feature is currently undergoing maintenance. Please retry in one hour.")
		return errcode.NewInvalidInputErr(err)
	}
	return nil
}

func (s *dmService) CreateJob(ctx context.Context, req *dmpb.CreateJobReq) (resp *dmpb.CreateJobResp, rErr error) {
	var sourceConnProfiles []entity.DFConnProfile
	defer func() {
		s.recordAuditLogForCreateJob(ctx, req, resp, sourceConnProfiles, rErr)
	}()
	resp = &dmpb.CreateJobResp{}
	userCtx, logger, cluster, rErr := s.checkBasic(ctx, req.ProjectId, req.ClusterId)
	if rErr != nil {
		return nil, rErr
	}

	if !bizutils.ClusterAvailable(cluster) {
		return nil, errcode.NewInvalidInputErr(errors.New("cluster is not in normal or scaling status"))
	}

	sourceConnProfileIDs := []entity.ID{}
	for i := range req.Sources {
		if req.FullInstanceMigrate && req.Sources[i].BaRules != nil {
			return nil, errcode.NewInvalidInputErr(errors.New("Cannot specify block/allow list when doing full instance migrate."))
		} else if !req.FullInstanceMigrate && req.Sources[i].BaRules == nil {
			return nil, errcode.NewInvalidInputErr(errors.New("Must specify block/allow list when not doing full instance migrate."))
		}
		if req.Sources[i].BaRules != nil && s.isEmptyBARules(req.Sources[i].BaRules) {
			return nil, errcode.NewInvalidInputErr(errors.New("Block/allow list is specified, but it's empty. Please set some objects to migrate."))
		}
		sourceConnProfileIDs = append(sourceConnProfileIDs, entity.ID(req.Sources[i].ConnProfileId))
	}
	sourceConnProfiles, rErr = connprofile.GetConnProfileByIDs(sourceConnProfileIDs...)
	if rErr != nil {
		logger.Error("get source conn profile failed", zap.Any("ids", sourceConnProfileIDs), zap.Error(rErr))
		return nil, errcode.NewInternalErr(errors.Errorf("failed to get source connection profile"))
	}
	if len(sourceConnProfiles) != len(sourceConnProfileIDs) {
		notFoundConnProfileIDs := []uint64{}
		for _, id := range sourceConnProfileIDs {
			found := false
			for _, profile := range sourceConnProfiles {
				if profile.ID == id {
					found = true
					break
				}
			}
			if !found {
				notFoundConnProfileIDs = append(notFoundConnProfileIDs, uint64(id))
			}
		}
		logger.Error("get connection profile failed for ids", zap.Uint64s("ids", notFoundConnProfileIDs))
		return nil, errcode.NewNotFoundErr(errors.Errorf("cannot find connection profiles for %+v", notFoundConnProfileIDs))
	}
	for _, p := range sourceConnProfiles {
		if p.ClusterID != entity.ID(cluster.Id) {
			return nil, errcode.NewForbiddenErr(errors.New("Cluster id of connection profile mismatch with input cluster id."))
		}
	}
	_, rErr = dm.GetJobSpecByID(entity.ID(req.SpecId))
	if rErr != nil {
		logger.Error("get job spec failed", zap.Uint64("spec-id", req.SpecId), zap.Error(rErr))
		if errors.Cause(rErr) == gorm.ErrRecordNotFound {
			return nil, errcode.NewInvalidInputErr(errors.New("Invalid job spec."))
		}
		return nil, errcode.NewInternalErr(errors.Errorf("Failed to get job spec."))
	}
	if err := s.checkQuota(ctx, logger, req.OrgId); err != nil {
		return nil, err
	}
	precheck, err := dm.GetPrecheckByID(entity.ID(req.PrecheckId))
	if err != nil {
		logger.Error("get job precheck failed", zap.Uint64("precheck-id", req.PrecheckId), zap.Error(err))
		if errors.Cause(err) == gorm.ErrRecordNotFound {
			return nil, errcode.NewInvalidInputErr(errors.New("Invalid precheck id."))
		}
		return nil, errcode.NewInternalErr(errors.Errorf("Failed to get precheck info."))
	}
	// we can only check at cluster level, since pre-check happens before create job, there's no strong relation between
	// them, so we cannot be sure 100% whether this pre-check belongs to this job.
	if precheck.ClusterID.Uint64() != req.ClusterId {
		return nil, errcode.NewInvalidInputErr(errors.New("Invalid precheck id."))
	}

	checkJob := &entity.DFJob{
		TenantID:  entity.ID(cluster.TenantId),
		ProjectID: entity.ID(cluster.ProjectId),
		ClusterID: entity.ID(cluster.Id),
		Region:    cluster.RegionName,
		Provider:  cluster.InfraProviderType,
	}
	if err = s.checkMaintenance(checkJob, logger, "create job"); err != nil {
		return nil, err
	}

	resp.Id, err = userCtx.Create(ctx, sourceConnProfiles, req, cluster)

	return resp, err
}

func (s *dmService) recordAuditLogForCreateJob(ctx context.Context, req *dmpb.CreateJobReq, resp *dmpb.CreateJobResp, sourceConnProfiles []entity.DFConnProfile, rErr error) {
	logger := log.WithContextS(ctx)
	cluster, bizError := grpcli.GetClusterByIDAndStatuses(ctx, req.ClusterId, nil)
	if bizError != nil {
		logger.Warn("failed to submit audit log, query cluster error", zap.Uint64("clusterId", req.ClusterId), zap.Error(bizError))
		return
	}

	details := auditlog.Details{
		auditlog.KeyJobName:           req.Name,
		auditlog.KeyTargetRegion:      cluster.RegionName,
		auditlog.KeyTargetClusterId:   cluster.Id,
		auditlog.KeyTargetClusterName: cluster.Name,
		auditlog.KeyMigrationJobType:  req.Mode,
		auditlog.KeyTargetUsername:    req.ClusterUser,
		auditlog.KeyError:             rErr,
	}
	if resp != nil && resp.Id != 0 {
		details[auditlog.KeyMigrationId] = resp.Id
	}
	if len(sourceConnProfiles) > 0 {
		details[auditlog.KeySourceDataSource] = sourceConnProfiles[0].Type
		details[auditlog.KeySourceRegion] = sourceConnProfiles[0].Region
		details[auditlog.KeySourceConnectivityMethod] = sourceConnProfiles[0].ConnMethod
		details[auditlog.KeySourceHostnameOrIpaddress] = sourceConnProfiles[0].Host
		details[auditlog.KeySourcePort] = sourceConnProfiles[0].Port
		details[auditlog.KeySourceUsername] = sourceConnProfiles[0].User
		if sourceConnProfiles[0].SslKey != "" || sourceConnProfiles[0].SslCert != "" || sourceConnProfiles[0].SslCa != "" {
			details[auditlog.KeySourceEnableTls] = true
		} else {
			details[auditlog.KeySourceEnableTls] = false
		}
	}
	err := auditlog.NewEvent(ctx, auditlog.EventCreateMigrationJob).
		SetResult(rErr == nil).
		SetOrgID(req.OrgId).
		SetProjectID(req.ProjectId).
		SetDetails(details.Build()).
		Submit()
	if err != nil {
		logger.Warn("failed to submit audit log", zap.Error(err))
		return
	}
}

func (s *dmService) isEmptyBARules(rules *dmpb.BlockAllowRules) bool {
	return len(rules.DoDbs) == 0 && len(rules.DoTables) == 0 &&
		len(rules.IgnoreDbs) == 0 && len(rules.IgnoreTables) == 0
}

type jobAPIParam struct {
	userCtx dm.UserContext
	logger  *zap.Logger
	dfJob   *entity.DFJob
}

func (s *dmService) checkBasic(ctx context.Context, projectID, clusterID uint64) (
	*dm.UserContext,
	*zap.Logger,
	*core.Cluster,
	error,
) {
	var (
		cluster *core.Cluster
		err     error
	)
	userCtx := s.userCtx(ctx)
	logger := userCtx.Log()
	if err = userCtx.CheckProjectPermission(entity2.ID(projectID)); err != nil {
		return nil, nil, nil, err
	}
	// user has access to target cluster
	cluster, err = grpcli.GetClusterByID(ctx, clusterID)
	if err != nil {
		logger.Error("get target cluster failed", zap.Error(err))
		// todo: difference between not-found and internal error
		return nil, nil, nil, errcode.NewInvalidInputErr(errors.New("Failed to get cluster information. Please check whether the cluster is cleared or deleted."))
	}
	if !feature.IsFeatureEnabledByCluster(feature.DM, cluster) {
		return nil, nil, nil, errcode.NewInvalidInputErr(errors.New("Data migration is not enabled on this cluster, please check whether this cluster meets the prerequisites in the doc."))
	}
	if cluster.ProjectId != projectID {
		return nil, nil, nil, errcode.NewInvalidInputErr(errors.New("project id mismatch with cluster's project id"))
	}
	return &userCtx, logger, cluster, nil
}

func (s *dmService) checkQuota(ctx context.Context, logger *zap.Logger, tenantID uint64) error {
	if activeJobs, err := dm.GetActiveDFJobByTenant(entity.ID(tenantID)); err != nil {
		logger.Error("fail to get active jobs", zap.Error(err))
		return errcode.NewInternalErr(errors.New("failed to get active jobs"))
	} else if len(activeJobs) >= feature.GetTenantQuota(tenantID) {
		logger.Warn("unable to create extra jobs", zap.Reflect("jobs", activeJobs), zap.Uint64("tenant id", tenantID))
		return errcode.NewInvalidInputErr(errors.New("Data Migration job count quota exceed in your tenant"))
	}
	return nil
}

func (s *dmService) checkJob(ctx context.Context, projectID, clusterID, jobID uint64) (*jobAPIParam, error) {
	userCtx := s.userCtx(ctx)
	logger := userCtx.Log().With(zap.Uint64("job-id", jobID))
	if err := userCtx.CheckProjectPermission(entity2.ID(projectID)); err != nil {
		return nil, err
	}
	dfJob, err := dm.GetDFJobByID(entity.ID(jobID))
	if err != nil {
		if errors.Cause(err) == gorm.ErrRecordNotFound {
			errMsg := fmt.Sprintf("Not found dataflow job of id %d", jobID)
			logger.Error(errMsg, zap.Error(err))
			return nil, errcode.NewNotFoundErr(errors.New(errMsg))
		}
		errMsg := fmt.Sprintf("Failed to get dataflow job of id %d", jobID)
		logger.Error(errMsg, zap.Error(err))
		return nil, errcode.NewInternalErr(errors.New(errMsg))
	}
	if dfJob.ServiceCode != entity.ServiceCodeDM {
		return nil, errcode.NewInvalidInputErr(errors.New("The job with requested id is not a Data Migration job."))
	}
	if dfJob.ProjectID != entity.ID(projectID) {
		return nil, errcode.NewInvalidInputErr(errors.New("project id mismatch with job's project id"))
	}
	if dfJob.ClusterID != entity.ID(clusterID) {
		errMsg := fmt.Sprintf("Dataflow job's cluster id mismatches with request's cluster id %d", clusterID)
		logger.Error(errMsg, zap.Reflect("dataflow job", dfJob), zap.Error(err))
		return nil, errcode.NewForbiddenErr(errors.New(errMsg))
	}

	return &jobAPIParam{
		userCtx: userCtx,
		logger:  logger,
		dfJob:   dfJob,
	}, nil
}

func (s *dmService) GetJob(ctx context.Context, req *dmpb.GetJobReq) (*dmpb.GetJobResp, error) {
	param, err := s.checkJob(ctx, req.ProjectId, req.ClusterId, req.Id)
	if err != nil {
		return nil, err
	}

	dfJob, logger := param.dfJob, param.logger

	if err = s.checkMaintenance(dfJob, logger, "list jobs"); err != nil {
		return nil, err
	}

	resp := &dmpb.GetJobResp{
		Job: &dmpb.Job{
			Id:         dfJob.ID.Uint64(),
			Name:       dfJob.Name,
			Region:     dfJob.Region,
			CreateTime: dfJob.CreatedAt.Format(time.RFC3339),
		},
	}

	if err := getDMJob(ctx, logger, dfJob, resp.Job); err != nil {
		return nil, err
	}

	return resp, nil
}

func (s *dmService) ListJobs(ctx context.Context, req *dmpb.ListJobsReq) (*dmpb.ListJobsResp, error) {
	_, logger, cluster, err := s.checkBasic(ctx, req.ProjectId, req.ClusterId)
	if err != nil {
		return nil, err
	}

	pager := utils.AdjustedPager(&req.Page, &req.PageSize)
	total, dfJobs, err := dm.ListDFJobsOfDM(entity.ID(cluster.Id), pager)
	if err != nil {
		return nil, err
	}

	resp := &dmpb.ListJobsResp{
		Total: uint64(total),
		Jobs:  make([]*dmpb.Job, len(dfJobs)),
	}

	var wg sync.WaitGroup
	for i, dfJob := range dfJobs {
		if err = s.checkMaintenance(&dfJob, logger, "list jobs"); err != nil {
			return nil, err
		}

		job := &dmpb.Job{
			Id:         dfJob.ID.Uint64(),
			Name:       dfJob.Name,
			Region:     dfJob.Region,
			CreateTime: dfJob.CreatedAt.Format(time.RFC3339),
		}
		resp.Jobs[i] = job

		wg.Add(1)
		go func() {
			defer func() {
				wg.Done()
			}()
			if err := getDMJob(ctx, logger.With(zap.Uint64("job-id", job.Id)), &dfJob, job); err != nil {
				logger.Error("failed to get dm job detail", zap.Uint64("job-id", job.Id), zap.Error(err))
			}
		}()
	}
	wg.Wait()

	return resp, nil
}

func (s *dmService) UpdateJob(ctx context.Context, req *dmpb.UpdateJobReq) (*emptypb.Empty, error) {
	param, err := s.checkJob(ctx, req.ProjectId, req.ClusterId, req.Id)
	if err != nil {
		return nil, err
	}

	if param.dfJob.Phase == entity.DFJobPhaseDeleting {
		return nil, errcode.NewInvalidInputErr(errors.New("Job is deleting, cannot update it."))
	}

	if err = s.checkMaintenance(param.dfJob, param.logger, "update job"); err != nil {
		return nil, err
	}

	if err = dm.UpdateDFJobName(param.dfJob, req.GetName()); err != nil {
		errMsg := fmt.Sprintf("Failed to update dataflow job's name to %s", req.GetName())
		param.logger.Error(errMsg, zap.Reflect("dataflow job", param.dfJob), zap.Error(err))
		return nil, errcode.NewInternalErr(errors.New(errMsg))
	}

	return &emptypb.Empty{}, nil
}

func (s *dmService) DeleteJob(ctx context.Context, req *dmpb.DeleteJobReq) (_ *emptypb.Empty, rErr error) {
	param, rErr := s.checkJob(ctx, req.ProjectId, req.ClusterId, req.Id)
	defer func() {
		s.recordAuditLogForDeleteJob(ctx, req, param, rErr)
	}()
	if rErr != nil {
		return nil, rErr
	}

	if err := s.checkMaintenance(param.dfJob, param.logger, "update job"); err != nil {
		return nil, err
	}

	if rErr = dm.UpdateDFJobPhase(param.dfJob, entity.DFJobPhaseDeleting); rErr != nil {
		msg := "failed to delete job"
		param.logger.Error(msg, zap.Error(rErr))
		return nil, errcode.NewInternalErr(errors.New(msg))
	}
	param.dfJob.Phase = entity.DFJobPhaseDeleting
	dm.PublishEvent(ctx, &dm.PublishEventReq{Job: &dm.Job{DFJob: param.dfJob}})

	return &emptypb.Empty{}, nil
}

func (s *dmService) recordAuditLogForDeleteJob(ctx context.Context, req *dmpb.DeleteJobReq, param *jobAPIParam, rErr error) {
	logger := log.WithContextS(ctx)
	details := auditlog.Details{
		auditlog.KeyMigrationId: req.Id,
		auditlog.KeyError:       rErr,
	}
	if param != nil && param.dfJob != nil && param.dfJob.Name != "" {
		details[auditlog.KeyJobName] = param.dfJob.Name
	}
	err := auditlog.NewEvent(ctx, auditlog.EventDeleteMigrationJob).
		SetResult(rErr == nil).
		SetOrgID(req.OrgId).
		SetProjectID(req.ProjectId).
		SetDetails(details.Build()).
		Submit()
	if err != nil {
		logger.Warn("failed to submit audit log", zap.Error(err))
		return
	}
}

func (s *dmService) OperateJob(ctx context.Context, req *dmpb.OperateJobReq) (_ *emptypb.Empty, rErr error) {
	param, rErr := s.checkJob(ctx, req.ProjectId, req.ClusterId, req.Id)
	defer func() {
		s.recordAuditLogForOperateJob(ctx, req, param, rErr)
	}()
	if rErr != nil {
		return nil, rErr
	}

	if param.dfJob.Phase != entity.DFJobPhaseCreated {
		errMsg := fmt.Sprintf("Job is in %s phase, cannot do %s on the job.", param.dfJob.Phase, req.Op)
		param.logger.Error(errMsg)
		return nil, errcode.NewInvalidInputErr(errors.New(errMsg))
	}

	if err := s.checkMaintenance(param.dfJob, param.logger, "update job"); err != nil {
		return nil, err
	}

	rErr = param.userCtx.OperateDMJob(ctx, param.dfJob, req)

	if rErr != nil {
		return nil, rErr
	}
	return &emptypb.Empty{}, nil
}

func (s *dmService) recordAuditLogForOperateJob(ctx context.Context, req *dmpb.OperateJobReq, param *jobAPIParam, rErr error) {
	logger := log.WithContextS(ctx)
	details := auditlog.Details{
		auditlog.KeyMigrationId: req.Id,
		auditlog.KeyError:       rErr,
	}
	if param != nil && param.dfJob != nil && param.dfJob.Name != "" {
		details[auditlog.KeyJobName] = param.dfJob.Name
	}

	var eventType auditlog.EventType = -1
	switch req.Op {
	case dmpb.OperateJobReq_Pause:
		eventType = auditlog.EventSuspendMigrationJob
	case dmpb.OperateJobReq_Resume:
		eventType = auditlog.EventResumeMigrationJob
	}
	if eventType < 0 {
		logger.Warn("failed to submit audit log, do support Operator type")
		return
	}

	err := auditlog.NewEvent(ctx, eventType).
		SetResult(rErr == nil).
		SetOrgID(req.OrgId).
		SetProjectID(req.ProjectId).
		SetDetails(details.Build()).
		Submit()
	if err != nil {
		logger.Warn("failed to submit audit log", zap.Error(err))
		return
	}
}

func (s *dmService) CancelPrecheck(ctx context.Context, req *dmpb.CancelPrecheckReq) (*emptypb.Empty, error) {
	var (
		logger  *zap.Logger
		err     error
		userCtx *dm.UserContext
	)
	if userCtx, logger, _, err = s.checkBasic(ctx, req.GetProjectId(), req.GetClusterId()); err != nil {
		return nil, err
	}
	if precheckJob, err := userCtx.GetPrecheckByID(entity.ID(req.Id)); err != nil {
		if errors.Cause(err) == gorm.ErrRecordNotFound {
			return nil, errcode.NewNotFoundErr(errors.Annotate(err, "precheck job with the input id not exists"))
		}
		logger.Error("fail to get precheck job by id", zap.Error(err))
		return nil, errcode.NewInternalErr(errors.Annotate(err, "fail to get precheck job by id"))
	} else if precheckJob.ClusterID != entity.ID(req.ClusterId) {
		logger.Error("precheck job id does not match the cluster id", zap.Uint64("expected cluster id", uint64(precheckJob.ClusterID)),
			zap.Uint64("input cluster id", req.ClusterId))
		return nil, errcode.NewInvalidInputErr(errors.New("precheck job id does not match the cluster id"))
	} else if precheckJob.Status != entity.PrecheckStatusRunning {
		logger.Warn("status of precheck job has been finalized")
		return &emptypb.Empty{}, nil
	}
	if err := dm.CancelPrecheck(entity.ID(req.Id)); err != nil {
		logger.Error("fail to cancel precheck", zap.Error(err))
		return nil, errcode.NewInternalErr(errors.Annotate(err, "fail to cancel precheck"))
	}
	return &emptypb.Empty{}, nil
}

func (s *dmService) GetQuota(ctx context.Context, req *dmpb.GetQuotaReq) (*dmpb.GetQuotaResp, error) {
	var err error

	if _, _, _, err = s.checkBasic(ctx, req.GetProjectId(), req.GetClusterId()); err != nil {
		return nil, err
	}
	resp := &dmpb.GetQuotaResp{
		Quota: int32(feature.GetTenantQuota(req.OrgId)),
	}
	activeJobs, err := dm.GetActiveDFJobByTenant(entity.ID(req.OrgId))
	if err != nil {
		return nil, errcode.NewInternalErr(errors.New("fail to get active dm jobs"))
	}
	resp.Usage = int32(len(activeJobs))
	return resp, nil
}

func (s *dmService) GetPrecheck(ctx context.Context, req *dmpb.GetPrecheckJobReq) (*dmpb.GetPrecheckJobResp, error) {
	var (
		logger                                      *zap.Logger
		userCtx                                     *dm.UserContext
		err                                         error
		finishedCnt, successCnt, warnCnt, failedCnt = 0, 0, 0, 0
	)
	resp := &dmpb.GetPrecheckJobResp{}
	if userCtx, logger, _, err = s.checkBasic(ctx, req.GetProjectId(), req.GetClusterId()); err != nil {
		return nil, err
	}
	precheckJob, err := userCtx.GetPrecheckByID(entity.ID(req.Id))
	if err != nil && errors.Cause(err) == gorm.ErrRecordNotFound {
		logger.Error("record not exists", zap.Error(err))
		return nil, errcode.NewNotFoundErr(errors.New("precheck job not exists"))
	} else if err != nil {
		logger.Error("fail to get precheck job", zap.Error(err))
		return nil, errcode.NewInternalErr(errors.Annotate(err, "fail to get precheck job"))
	}
	if precheckJob.ClusterID != entity.ID(req.ClusterId) {
		logger.Error("unmatched cluster id")
		return nil, errcode.NewInvalidInputErr(errors.New("precheck job's cluster id not matches the input cluster id"))
	}
	resp.Status = string(precheckJob.Status)
	resp.Total = int32(precheckJob.TotalItem)
	if precheckJob.Status == entity.PrecheckStatusCancel {
		return resp, nil
	}
	items, err := dm.GetPrecheckItemsByJobID(entity.ID(req.Id))
	if err != nil {
		logger.Error("fail to get precheck items", zap.Error(err))
		return nil, errcode.NewInternalErr(errors.Annotate(err, "fail to get precheck items"))
	}
	for _, item := range items {
		logger.Debug("precheck item", zap.Reflect("item", item))
		if item.Status == entity.PrecheckStatusFailed {
			// internal error
			resp.Items = append(resp.Items, &dmpb.PrecheckItem{
				Desc:           item.Desc,
				Status:         string(checker.StateFailure),
				Reason:         item.ErrMsg, // TODO: refine internal error message
				Solution:       fmt.Sprintf("Precheck(id = %d) failed, please contact the support team", req.Id),
				SolutionDocUrl: dm.GetDmDocUrl(),
			})
			failedCnt++
			finishedCnt++
			continue
		}
		if item.Status == entity.PrecheckStatusRunning {
			continue
		}
		respItem := formatCheckItemResult(item)
		if respItem == nil {
			// should not happen
			continue
		}
		resp.Items = append(resp.Items, respItem)
		switch respItem.Status {
		case string(checker.StateFailure):
			failedCnt++
		case string(checker.StateWarning):
			warnCnt++
		case string(checker.StateSuccess):
			successCnt++
		}
		finishedCnt++
	}
	resp.FailedCnt = int32(failedCnt)
	resp.WarnCnt = int32(warnCnt)
	resp.SuccessCnt = int32(successCnt)
	if finishedCnt < precheckJob.TotalItem {
		resp.Status = string(entity.PrecheckStatusRunning)
	} else if failedCnt > 0 {
		resp.Status = string(entity.PrecheckStatusFailed)
	} else {
		resp.Status = string(entity.PrecheckStatusFinished)
	}
	return resp, nil
}

// ListPrechecks lists the precheck records of a cluster (reverse chronological).
// Per-item details are not included; callers fetch them via GetPrecheck.
func (s *dmService) ListPrechecks(ctx context.Context, req *dmpb.ListPrechecksReq) (*dmpb.ListPrechecksResp, error) {
	if _, _, _, err := s.checkBasic(ctx, req.GetProjectId(), req.GetClusterId()); err != nil {
		return nil, err
	}
	pager := utils.AdjustedPager(&req.Page, &req.PageSize)
	total, prechecks, err := dm.ListPrechecksOfCluster(entity.ID(req.GetClusterId()), pager)
	if err != nil {
		return nil, errcode.NewInternalErr(errors.Annotate(err, "fail to list prechecks"))
	}
	if len(prechecks) == 0 {
		return &dmpb.ListPrechecksResp{Total: uint64(total)}, nil
	}
	ids := make([]entity.ID, 0, len(prechecks))
	for i := range prechecks {
		ids = append(ids, prechecks[i].ID)
	}
	itemsByPrecheck, err := dm.ListPrecheckItemsByPrecheckIDs(ids)
	if err != nil {
		return nil, errcode.NewInternalErr(errors.Annotate(err, "fail to load precheck items"))
	}
	resp := &dmpb.ListPrechecksResp{
		Total:     uint64(total),
		Prechecks: make([]*dmpb.PrecheckSummary, 0, len(prechecks)),
	}
	for i := range prechecks {
		p := &prechecks[i]
		summary := &dmpb.PrecheckSummary{
			PrecheckId:          p.ID.Uint64(),
			CreatedAt:           timestamppb.New(p.CreatedAt),
			Status:              string(p.Status),
			Total:               int32(p.TotalItem),
			FullInstanceMigrate: p.FullInstanceMigrate,
		}
		summary.FailedCnt, summary.WarnCnt, summary.SuccessCnt = aggregatePrecheckCounts(itemsByPrecheck[p.ID])
		resp.Prechecks = append(resp.Prechecks, summary)
	}
	return resp, nil
}

// aggregatePrecheckCounts mirrors the aggregation logic of GetPrecheck.
// Running items are skipped; failed-status items count toward failed.
func aggregatePrecheckCounts(items []entity.DMPrecheckItem) (failed, warn, success int32) {
	for i := range items {
		item := items[i]
		if item.Status == entity.PrecheckStatusRunning {
			continue
		}
		if item.Status == entity.PrecheckStatusFailed {
			failed++
			continue
		}
		respItem := formatCheckItemResult(item)
		if respItem == nil {
			continue
		}
		switch respItem.Status {
		case string(checker.StateFailure):
			failed++
		case string(checker.StateWarning):
			warn++
		case string(checker.StateSuccess):
			success++
		}
	}
	return
}

func (s *dmService) PrecheckJob(ctx context.Context, req *dmpb.CreateJobReq) (*dmpb.CreateJobResp, error) {
	var (
		resp    = &dmpb.CreateJobResp{}
		userCtx *dm.UserContext
		logger  *zap.Logger
		err     error
	)
	if userCtx, logger, _, err = s.checkBasic(ctx, req.GetProjectId(), req.GetClusterId()); err != nil {
		return nil, err
	}
	if len(req.Sources) == 0 || req.Sources[0].ConnProfile == nil {
		logger.Error("source config is empty")
		return nil, errcode.NewInvalidInputErr(errors.New("source config is empty"))
	}
	connProfile := req.Sources[0].ConnProfile
	if (connProfile.EndpointServiceName != nil && connProfile.Host != nil) ||
		(connProfile.EndpointServiceName == nil && connProfile.Host == nil) {
		msg := "invalid conn profile, either service name or host should be set"
		logger.Error(msg)
		return nil, errcode.NewInvalidInputErr(errors.New(msg))
	}
	if connProfile.ConnMethod == string(entity.ConnMethodPrivateLink) {
		if connProfile.EndpointServiceName == nil {
			return nil, errcode.NewInvalidInputErr(errors.New("Endpoint service name should not be empty when using private link"))
		}
	} else if connProfile.Host == nil {
		return nil, errcode.NewInvalidInputErr(errors.New("Host should not be empty when using Public IP or VPC Peering."))
	}

	if err = utils.CheckTLSParam(connProfile.SslCert, connProfile.SslKey); err != nil {
		return nil, err
	}
	precheckID, err := userCtx.CreatePrecheck(req)
	if err != nil {
		logger.Error("create precheck failed", zap.Error(err))
		// TODO: refine error message
		return nil, errcode.NewInternalErr(errors.New("fail to create precheck job"))
	}
	resp.Id = uint64(precheckID)
	// TODO: timeout?
	go func() {
		defer func() {
			if err := recover(); err != nil {
				logger.Error("precheck job panic", zap.Any("cause", err))
				// update precheck job status
				precheckJob, err1 := userCtx.GetPrecheckByID(precheckID)
				if err1 != nil {
					logger.Error("fail to get precheck job after panic", zap.Error(err1))
					return
				}
				precheckJob.Status = entity.PrecheckStatusFailed
				logger.Debug("write precheck result", zap.Reflect("result", precheckJob))
				err2 := userCtx.UpdatePrecheck(precheckJob)
				if err2 != nil {
					logger.Error("fail to update precheck job after panic", zap.Error(err2))
				}
			}
		}()
		userCtx.PrecheckWorker(context.Background(), req, precheckID, logid.GetLogid(ctx))
	}()
	return resp, nil
}

func (s *dmService) ScaleUpDownJob(ctx context.Context, req *dmpb.ScaleUpDownJobReq) (*emptypb.Empty, error) {
	var (
		param *jobAPIParam
		err   error
	)
	if param, err = s.checkJob(ctx, req.GetProjectId(), req.GetClusterId(), req.GetId()); err != nil {
		return &emptypb.Empty{}, err
	}
	logger := s.model.Log().With(log.FieldsFromContext(ctx)...)
	return &emptypb.Empty{}, s.model.ModifyJob(ctx, logger, &param.userCtx, param.dfJob, entity.ID(req.Id), req.ToSpecId)
}

func (s *dmService) GetSourceTables(ctx context.Context, req *dataflow.ValidateRemoteDBConnectivityReq) (*agent.GetSourceTablesResp, error) {
	_, logger, _, err := s.checkBasic(ctx, req.GetProjectId(), req.GetClusterId())
	if err != nil {
		return nil, err
	}
	cluster, _, agentClient, err := common.GetClusterNetworkAgentById(ctx, req.ClusterId)
	if err != nil {
		return nil, errors.Annotate(err, "get cluster, network, agent client failed")
	}

	var host string
	if entity.ConnMethod(req.ConnMethod) == entity.ConnMethodPrivateLink {
		if req.EndpointServiceName == nil {
			return nil, errcode.NewInvalidInputErr(errors.New("Endpoint service name should not be empty when using private link"))
		}
		endpointServiceName := strings.TrimSpace(*req.EndpointServiceName)
		privateLink, err2 := connprofile.GetAvailablePrivateLink(cluster.GetDataPlaneInfo().GetK8SNamespace(), cluster.GetDataPlaneInfo().GetNetworkName(), endpointServiceName)
		if err2 != nil {
			logger.Error("get private link info failed", zap.Error(err2))
			return nil, errcode.NewInternalErr(errors.New("get private link failed"))
		}
		host = privateLink.RegionalDNS
	} else {
		if req.Host == nil {
			return nil, errcode.NewInvalidInputErr(errors.New("Host should not be empty when using Public IP or VPC Peering."))
		}
		host = strings.TrimSpace(*req.Host)
	}

	if err = utils.CheckTLSParam(req.SslCert, req.SslKey); err != nil {
		return nil, err
	}

	resp, err := agentClient.GetSourceTables(ctx, &agent.CheckDBConnectivityReq{
		Host:      host,
		Port:      req.Port,
		User:      req.User,
		Password:  req.Password,
		SslCa:     req.SslCa,
		SslCert:   req.SslCert,
		SslKey:    req.SslKey,
		RequestId: logid.GetLogid(ctx),
	})
	if err != nil {
		logger.Error("failed to get all tables", zap.Error(err))
		return nil, errcode.NewInternalErr(errors.New("failed to get all tables"))
	}
	return resp, nil
}

func (s *dmService) GetCost(ctx context.Context, req *dmpb.GetCostReq) (*dmpb.GetCostResp, error) {
	_, logger, cluster, err := s.checkBasic(ctx, req.GetProjectId(), req.GetClusterId())
	if err != nil {
		return nil, err
	}

	jobSpec, err := dm.GetJobSpecByID(entity.ID(req.SpecId))
	if err != nil {
		logger.Error("get job spec failed", zap.Uint64("spec-id", req.SpecId), zap.Error(err))
		if errors.Cause(err) == gorm.ErrRecordNotFound {
			return nil, errcode.NewInvalidInputErr(errors.New("Invalid job spec."))
		}
		return nil, errcode.NewInternalErr(errors.Errorf("Failed to get job spec."))
	}
	nodeSKU, err := commondm.GetNodeRCUPrice(cluster.CloudProviderName, cluster.RegionName)
	if err != nil {
		logger.Error("get node sku failed", zap.String("cluster-provider", cluster.CloudProviderName),
			zap.String("region", cluster.RegionName), zap.Error(err))
		return nil, errcode.NewInvalidInputErr(errors.Errorf("Failed to get cost."))
	}
	// after discussion, private link endpoint is not charged now
	// see https://github.com/tidbcloud/cluster-tier-config/blob/main/nightly/sku-dm.yaml (dev)
	return &dmpb.GetCostResp{
		NodeCost: nodeSKU * jobSpec.CapacityUnit,
	}, nil
}

func getDMJob(ctx context.Context, logger *zap.Logger, dfJob *entity.DFJob, job *dmpb.Job) error {
	dmJob, err := dm.GetDMJobByID(dfJob.ID)
	if err != nil {
		if err == gorm.ErrRecordNotFound {
			errMsg := fmt.Sprintf("Not found dm job %s", dfJob.Name)
			logger.Error("Not found dm job by job id", zap.Error(err))
			return errcode.NewNotFoundErr(errors.New(errMsg))
		}
		errMsg := fmt.Sprintf("Failed to access DM job %s, please contact our support team.", dfJob.Name)
		logger.Error(errMsg, zap.Error(err))
		return errcode.NewInternalErr(errors.New(errMsg))
	}
	job.ClusterUser = dmJob.ClusterUser
	job.Mode = string(dmJob.Mode)
	job.ImportMode = string(dmJob.ImportMode)

	jobSpec, err := dm.GetJobSpecByID(dmJob.SpecID)
	if err != nil {
		logger.Error("failed to get job spec", zap.Error(err))
		return errcode.NewInternalErr(errors.New("Failed to get job detail"))
	}
	job.Spec = jobSpecEntity2PB(jobSpec)

	dmJobSources, err := dm.GetDMJobSources(dfJob.ID)
	if err != nil {
		errMsg := "Failed to get job, please contact our support team."
		logger.Error("failed to get dm job sources", zap.Error(err))
		return errcode.NewInternalErr(errors.New(errMsg))
	}
	var currentStep dmpb.SubTask_Step
	job.SubTasks = make(map[uint64]*dmpb.SubTask, len(dmJobSources))
	for _, dmJobSource := range dmJobSources {
		connProfileID := dmJobSource.ConnProfileID.Uint64()
		subTask := &dmpb.SubTask{}
		job.SubTasks[connProfileID] = subTask

		source := jobSource2Dataflow(dmJobSource)
		subTask.Source = source
		connProfileEntity, err := connprofile.MustGetConnProfileByID(entity.ID(connProfileID))
		if err != nil {
			errMsg := "Failed to get job detail, please contact our support team."
			logger.Error("failed to get connection profile", zap.Error(err), zap.Uint64("conn-profile-id", source.ConnProfileId))
			return errcode.NewInternalErr(errors.New(errMsg))
		}
		source.ConnProfile = connProfileEntity2Dataflow(connProfileEntity)

		switch dmJob.Mode {
		case entity.DMJobModeAll:
			if !dmJobSource.DumpStatus.IsFinished() {
				subTask.CurrentStep = dmpb.SubTask_Dump
			} else if !dmJobSource.LoadStatus.IsFinished() {
				subTask.CurrentStep = dmpb.SubTask_Load
			} else {
				subTask.CurrentStep = dmpb.SubTask_Sync
			}
		case entity.DMJobModeFull:
			if !dmJobSource.DumpStatus.IsFinished() {
				subTask.CurrentStep = dmpb.SubTask_Dump
			} else {
				subTask.CurrentStep = dmpb.SubTask_Load
			}
		case entity.DMJobModeIncr:
			subTask.CurrentStep = dmpb.SubTask_Sync
		}

		if dmJobSource.DumpStatus != nil {
			subTask.DumpDetail, err = dumpDetailEntity2PB(dmJobSource.DumpStatus)
			if err != nil {
				errMsg := "Failed to get job detail, please contact our support team."
				logger.Error("failed to get dump task detail", zap.Error(err), zap.Reflect("dump status", dmJobSource.DumpStatus))
				return errcode.NewInternalErr(errors.New(errMsg))
			}
		}

		if dmJobSource.LoadStatus != nil {
			subTask.LoadDetail, err = loadDetailEntity2PB(dmJobSource.LoadStatus)
			if err != nil {
				errMsg := "Failed to get job detail, please contact our support team."
				logger.Error("failed to get load task detail", zap.Error(err), zap.Reflect("dump status", dmJobSource.LoadStatus))
				return errcode.NewInternalErr(errors.New(errMsg))
			}
		}

		// If dfJob.Phase is created, this will be covered by genSubTaskStageByTask later
		subTask.Stage = genSubTaskStageByDFPhase(logger, dfJob.Phase)
		if subTask.Stage == dmpb.SubTask_Failed {
			if dfJob.ErrorMessage != "" {
				subTask.ProcessError = genProcessErrorFromDFJobError(logger, dfJob.ErrorMessage)
			} else {
				subTask.ProcessError = genUnknownError(logger)
			}
		}
		currentStep = subTask.CurrentStep
	}

	if dfJob.Phase != entity.DFJobPhaseCreated && dfJob.Phase != entity.DFJobPhaseModifying {
		logger.Info("dataflow job is not created", zap.String("df job name", dfJob.Name))
		return nil
	}

	if err := fillDMJobModification(ctx, logger, dfJob, job); err != nil {
		return err
	}

	dmOpenapi, err := dm.GetDMOpenapiClient(ctx, dfJob)
	if err != nil {
		return err
	}
	dmStage := dm.ConvertToStage(currentStep)
	taskName := dm.GenDMTaskName(dmJob.ID, dmStage)
	taskStatusResp, err := dmOpenapi.QueryTaskStatus(ctx, taskName, dm.GetDatasourceNames(dmJobSources))
	var taskStatus []openapi.SubTaskStatus
	if err == nil {
		taskStatus = taskStatusResp.Data
	} else if err == innerAgent.ErrorDMResourceNotFound {
		logger.Info("dm job task not found, may be not ready or already been deleted", zap.String("taskName", taskName))
		if dfJob.ErrorMessage != "" {
			for _, subTask := range job.SubTasks {
				if isProcessErrEmpty(subTask.ProcessError) {
					subTask.ProcessError = genProcessErrorFromDFJobError(logger, dfJob.ErrorMessage)
				}
			}
		}
	} else {
		errMsg := "Failed to query job status, please contact our support team."
		logger.Error("failed to query dm job task status", zap.String("taskName", taskName), zap.Error(err))
		return errcode.NewInternalErr(errors.New(errMsg))
	}
	if err = fillDMJobByTaskStatus(logger, job, taskStatus); err != nil {
		errMsg := "Internal error, please contact our support team."
		logger.Error("failed to fill dm job by task status", zap.String("taskName", taskName), zap.Error(err))
		return errcode.NewInternalErr(errors.New(errMsg))
	}
	if dfJob.Phase == entity.DFJobPhaseModifying {
		// mask the original stage of tasks
		for _, subTask := range job.SubTasks {
			subTask.Stage = dmpb.SubTask_Scaling
		}
	}
	return nil
}

func fillDMJobModification(_ context.Context, logger *zap.Logger, dfJob *entity.DFJob, job *dmpb.Job) error {
	if dfJob.Phase != entity.DFJobPhaseModifying {
		return nil
	}
	// get unfinished modifications
	modifications, err := dm.GetUnfinishedJobModification(dfJob.ID)
	if err != nil {
		logger.Error("fail to get unfinished job modification", zap.Error(err))
		return err
	}
	if len(modifications) == 0 {
		// uncommon case: the job might have been marked as created in other routine
		// and the all the modifications are finished.
		logger.Info("no unfinished job modification", zap.String("df job name", dfJob.Name))
	} else if len(modifications) > 1 {
		// should never happen
		logger.Warn("more than one unfinished job modification", zap.String("df job name", dfJob.Name))
	} else {
		mod := modifications[0]
		fromSpec, err := dm.GetJobSpecByID(mod.Param.FromSpecID)
		if err != nil {
			logger.Error("fail to get job spec", zap.Error(err))
			return err
		}
		toSpec, err := dm.GetJobSpecByID(mod.Param.ToSpecID)
		if err != nil {
			logger.Error("fail to get job spec", zap.Error(err))
			return err
		}
		job.Modification = &dmpb.JobModification{
			FromRcu: uint64(fromSpec.CapacityUnit),
			ToRcu:   uint64(toSpec.CapacityUnit),
		}
	}
	return nil
}

func formatCheckItemResult(item entity.DMPrecheckItem) (ret *dmpb.PrecheckItem) {
	raw := item.Results
	if len(raw.Results) == 0 {
		// running item
		// should not happen since we've checked it before.
		return nil
	}
	ret = &dmpb.PrecheckItem{
		Desc: raw.Results[0].Desc,
	}
	ret.Status = string(raw.Results[0].State)
	compactResult := raw.Results[0]
	for i := 1; i < len(raw.Results); i++ {
		switch raw.Results[i].State {
		case checker.StateFailure:
			ret.Status = string(checker.StateFailure)
		case checker.StateWarning:
			if ret.Status != string(checker.StateFailure) {
				ret.Status = string(checker.StateWarning)
			}
		case checker.StateSuccess:
			// do nothing because
			// 1. when compactResult is failure and warning, we should keep it
			// 2. when compactResult is success, we don't need to change it
		}
		compactResult.Errors = append(compactResult.Errors, raw.Results[i].Errors...)
		compactResult.Instruction += "; " + raw.Results[i].Instruction
	}
	// compact error message
	// TODO: when checking many tables' structures,
	// the error messages can possibly be a lot.
	for _, err := range compactResult.Errors {
		ret.Reason += err.ShortErr + "; "
	}
	ret.Solution = compactResult.Instruction
	if checker.State(ret.Status) != checker.StateSuccess {
		// the info stored in item.Desc is the item type actually
		ret.SolutionDocUrl = dm.GetPrecheckItemHelpUrl(item.Desc)
	}
	ret.Type = dm.GetPrecheckItemType(item.Desc)
	return ret
}