package dm

import (
	"bytes"
	"encoding/json"
	"fmt"
	"strconv"
	"strings"
	"time"

	"github.com/golang/protobuf/jsonpb"
	"github.com/pingcap/errcode"
	"github.com/pingcap/errors"
	"github.com/pingcap/tidb/pkg/util/filter"
	router "github.com/pingcap/tidb/pkg/util/table-router"
	"github.com/pingcap/tiflow/dm/openapi"
	tiflowdmpb "github.com/pingcap/tiflow/dm/pb"
	"github.com/pingcap/tiflow/dm/pkg/terror"
	engpb "github.com/pingcap/tiflow/engine/enginepb"
	frameModel "github.com/pingcap/tiflow/engine/framework/model"
	"github.com/pingcap/tiflow/engine/jobmaster/dm"
	"github.com/pingcap/tiflow/engine/jobmaster/dm/metadata"
	dmpkg "github.com/pingcap/tiflow/engine/pkg/dm"
	"github.com/tidbcloud/dataflow-service/idl/pbgen/proto/common"
	"github.com/tidbcloud/dataflow-service/idl/pbgen/proto/dataflow"
	dmpb "github.com/tidbcloud/dataflow-service/idl/pbgen/proto/dm"
	commondm "github.com/tidbcloud/dataflow-service/internal/common/dm"
	"github.com/tidbcloud/dataflow-service/internal/db/entity"
	dmModel "github.com/tidbcloud/dataflow-service/internal/model/dm"
	"go.uber.org/zap"
)

func EngineJob2Dataflow(log *zap.Logger, engJob *engpb.Job, job *dmpb.Job, dmJobSources []entity.DMJobSource) error {
	for _, subTask := range job.SubTasks {
		subTask.Stage = genSubTaskStateByJobState(log, engJob.State)
		if subTask.Stage == dmpb.SubTask_Failed {
			subTask.ProcessError = genUnknownError(log)
			return nil
		}
		if subTask.Stage == dmpb.SubTask_Deleting {
			errMsg := fmt.Sprintf("Not found dataflow job of id %d", job.Id)
			return errcode.NewNotFoundErr(errors.New(errMsg))
		}
	}

	var jobStatus *dm.JobStatus
	if len(engJob.Detail) != 0 {
		jobStatus = &dm.JobStatus{}
		if err := json.Unmarshal(engJob.Detail, jobStatus); err != nil {
			return err
		}
	} else {
		log.Info("engine job's detail is still nil")
	}

	for _, src := range dmJobSources {
		idStr := commondm.GenSourceID(src.ConnProfileID)
		connProfileId := src.ConnProfileID.Uint64()

		subTask, ok := job.SubTasks[connProfileId]
		if !ok {
			return errors.Errorf("job source(%d) mismatches dm subtask", connProfileId)
		}

		// seems engine always return engJob.Error, but its message might be empty
		// see https://github.com/pingcap/tiflow/blob/4dcf1396ff8fc9e20bd424e069195bf06442182e/engine/servermaster/jobmanager.go#L469-L471
		if engJob.Error != nil && len(engJob.Error.Message) != 0 {
			// such as:
			// {
			//    "id": "1111",
			//    "type": "DM",
			//    "state": "Created",
			//    "config": "",
			//    "detail": "",
			//    "error": {
			//        "code": "",
			//        "message": "Error 1044: Access denied for user 'xxx'@'%' to database 'tidbcloud_dm_meta'"
			//    },
			//}
			subTask.Stage = dmpb.SubTask_Failed
			subTask.ProcessError = &dmpb.ProcessError{
				Message: engJob.Error.Message,
			}
			// jobStatus should be nil in this case
			continue
		}

		var err error
		if jobStatus == nil {
			continue
		}

		if jobStatus.FinishedUnitStatus != nil {
			for _, finishedStatus := range jobStatus.FinishedUnitStatus[idStr] {
				if err = detail2PB(log, &dmpkg.QueryStatusResponse{
					Unit:   finishedStatus.Unit,
					Stage:  finishedStatus.Stage,
					Result: dmpkg.NewProcessResultFromPB(finishedStatus.Result),
					Status: finishedStatus.Status,
				}, subTask, finishedStatus.Duration); err != nil {
					return err
				}
			}
		}

		if jobStatus.TaskStatus != nil {
			taskStatus, ok := jobStatus.TaskStatus[idStr]
			if !ok {
				log.Warn("job status no such source", zap.Reflect("job status", jobStatus), zap.Reflect("source", src))
				return nil
			}
			if taskStatus.Status == nil {
				log.Info("QueryStatusResponse is still empty")
				return nil
			}
			status := taskStatus.Status

			subTask.Stage = genSubTaskStageByTask(log, status.Stage)
			if len(status.ErrorMsg) > 0 {
				log.Warn("some errors in query status", zap.String("error message", status.ErrorMsg), zap.Stringer("expected stage", taskStatus.ExpectedStage))
				// status.ErrorMsg existing does not mean job failed, so we use taskStatus.ExpectedStage to generate stage
				subTask.Stage = genSubTaskStageByTask(log, taskStatus.ExpectedStage)
			}
			if err = detail2PB(log, status, subTask, taskStatus.Duration); err != nil {
				return err
			}
			if subTask.Stage == dmpb.SubTask_Failed && isProcessErrEmpty(subTask.ProcessError) {
				subTask.ProcessError = genUnknownError(log)
			}
		}
	}
	return nil
}

func fillDMJobByTaskStatus(logger *zap.Logger, job *dmpb.Job, taskStatus []openapi.SubTaskStatus) error {

	if len(taskStatus) != len(job.SubTasks) {
		logger.Warn("dm subtask status not match with job source", zap.Int("subTaskLen", len(taskStatus)), zap.Int("jobSourceLen", len(job.SubTasks)))
	}

	subTaskStatusMap := make(map[uint64]*openapi.SubTaskStatus, len(taskStatus))
	for _, subTaskStatus := range taskStatus {
		_, connProfileID, err := dmModel.ParseDMDataSourceName(subTaskStatus.SourceName)
		if err != nil {
			logger.Error("invalid dm subtask source name",
				zap.Uint64("jobID", job.Id),
				zap.String("taskNmae", subTaskStatus.Name),
				zap.String("sourceNmae", subTaskStatus.SourceName))
			return err
		}
		subTaskStatusMap[connProfileID] = &subTaskStatus
	}

	for connProfileID, subTask := range job.SubTasks {
		subTaskStatus := subTaskStatusMap[connProfileID]
		if subTaskStatus == nil {
			logger.Warn("dm subtask not match with dm job source",
				zap.Uint64("jobID", job.Id),
				zap.Uint64("connProfileID", connProfileID))
			// subTask.Stage = dmpb.SubTask_Unknown
			// subTask.ProcessError = genUnknownError(logger)
			continue
		}
		if shouldIgnoreLiveSubTaskError(subTaskStatus) {
			logger.Warn("ignore live subtask error and keep stored status as fallback",
				zap.Uint64("jobID", job.Id),
				zap.Uint64("connProfileID", connProfileID),
				zap.Any("subTaskStatus", subTaskStatus))
			continue
		}
		taskStage := genSubTaskStateByDMSubTaskStage(logger, subTaskStatus)
		if taskStage != dmpb.SubTask_Finished {
			subTask.Stage = taskStage
		}
		if taskStage == dmpb.SubTask_Failed || taskStage == dmpb.SubTask_Unknown {
			subTask.ProcessError = genErrorWithSupport(logger, subTaskStatus.ErrorMsg)
		}

		switch subTaskStatus.Unit {
		case dmModel.DMTaskUintDump:
			subTask.CurrentStep = dmpb.SubTask_Dump
			if subTaskStatus.DumpStatus == nil {
				continue
			}
			progress, err := removePercent(subTaskStatus.DumpStatus.Progress)
			if err != nil {
				logger.Warn("invalid subtask load status progress",
					zap.Uint64("jobID", job.Id),
					zap.Uint64("connProfileID", connProfileID),
					zap.Any("loadStatus", subTaskStatus.LoadStatus),
				)
			}
			subTask.DumpDetail = &dmpb.DumpDetail{
				Bps:             subTaskStatus.DumpStatus.Bps,
				Progress:        progress,
				TotalTables:     subTaskStatus.DumpStatus.TotalTables,
				CompletedTables: int64(subTaskStatus.DumpStatus.CompletedTables),
				FinishedBytes:   int64(subTaskStatus.DumpStatus.FinishedBytes),
				FinishedRows:    int64(subTaskStatus.DumpStatus.FinishedRows),
			}

		case dmModel.DMTaskUintLoad:
			subTask.CurrentStep = dmpb.SubTask_Load
			if subTaskStatus.LoadStatus == nil {
				continue
			}
			progress, err := removePercent(subTaskStatus.LoadStatus.Progress)
			if err != nil {
				logger.Warn("invalid subtask load status progress",
					zap.Uint64("jobID", job.Id),
					zap.Uint64("connProfileID", connProfileID),
					zap.Any("loadStatus", subTaskStatus.LoadStatus),
				)
			}
			subTask.LoadDetail = &dmpb.LoadDetail{
				FinishedBytes: subTaskStatus.LoadStatus.FinishedBytes,
				TotalBytes:    subTaskStatus.LoadStatus.TotalBytes,
				Progress:      progress,
				Bps:           subTaskStatus.LoadStatus.Bps,
			}

		case dmModel.DMTaskUintSync:
			subTask.CurrentStep = dmpb.SubTask_Sync
			if subTaskStatus.SyncStatus == nil {
				continue
			}
			subTask.SyncDetail = &dmpb.SyncDetail{
				Rps:        subTaskStatus.SyncStatus.RecentTps,
				Latency:    subTaskStatus.SyncStatus.SecondsBehindMaster,
				Checkpoint: subTaskStatus.SyncStatus.SyncerBinlog,
			}

		default:
			// should not happen
			subTask.CurrentStep = dmpb.SubTask_Unsupported
			logger.Warn("an unsupported subtask type in dataflow service", zap.String("subtask type", subTaskStatus.Unit))
			return nil
		}
	}
	return nil
}

func shouldIgnoreLiveSubTaskError(subTaskStatus *openapi.SubTaskStatus) bool {
	if subTaskStatus == nil || subTaskStatus.ErrorMsg == nil || *subTaskStatus.ErrorMsg == "" {
		return false
	}
	if subTaskStatus.Stage != "" || subTaskStatus.Unit != "" {
		return false
	}

	errMsg := *subTaskStatus.ErrorMsg
	return strings.Contains(errMsg, "grpc request error") ||
		strings.Contains(errMsg, "DeadlineExceeded") ||
		strings.Contains(errMsg, "context deadline exceeded")
}

func routeRule2Dataflow(rule *router.TableRule) *dmpb.RouteRule {
	return &dmpb.RouteRule{
		SchemaPattern: rule.SchemaPattern,
		TablePattern:  rule.TablePattern,
		TargetSchema:  rule.TargetSchema,
		TargetTable:   rule.TargetTable,
	}
}

func baList2Dataflow(baList filter.Rules) *dmpb.BlockAllowRules {
	baRules := &dmpb.BlockAllowRules{
		DoDbs:     baList.DoDBs,
		IgnoreDbs: baList.IgnoreDBs,
	}

	baRules.DoTables = make([]*common.Table, 0, len(baList.DoTables))
	for _, table := range baList.DoTables {
		baRules.DoTables = append(baRules.DoTables, &common.Table{
			Schema: table.Schema,
			Table:  table.Name,
		})
	}

	baRules.IgnoreTables = make([]*common.Table, 0, len(baList.IgnoreTables))
	for _, table := range baList.IgnoreTables {
		baRules.IgnoreTables = append(baRules.IgnoreTables, &common.Table{
			Schema: table.Schema,
			Table:  table.Name,
		})
	}

	return baRules
}

func connProfileEntity2Dataflow(input *entity.DFConnProfile) *dataflow.ConnProfile {
	return &dataflow.ConnProfile{
		Id:                  input.ID.Uint64(),
		Name:                input.Name,
		Type:                input.Type.DisplayName(),
		Version:             input.Version,
		Region:              input.Region,
		Host:                &input.Host,
		Port:                uint32(input.Port),
		User:                input.User,
		ConnMethod:          input.ConnMethod.DisplayName(),
		EndpointServiceName: &input.EndpointServiceName,
	}
}

// genSubTaskStageByDFPhase is called when DFJobPhase is not `Deleted`.
func genSubTaskStageByDFPhase(logger *zap.Logger, status entity.DFJobPhase) dmpb.SubTask_Stage {
	switch status {
	case entity.DFJobPhaseCreating:
		return dmpb.SubTask_Creating
	case entity.DFJobPhaseCreated:
		return dmpb.SubTask_Running
	case entity.DFJobPhaseFinished, entity.DFJobPhaseFinishing:
		return dmpb.SubTask_Finished
	case entity.DFJobPhaseDeleting:
		return dmpb.SubTask_Deleting
	case entity.DFJobPhaseModifying:
		// there's only scale up/down right now, so map modifying to scaling directly.
		return dmpb.SubTask_Scaling
	}
	// this can't happen now
	logger.Warn("df job phase is unknown")
	return dmpb.SubTask_Failed
}

// genSubTaskStageByTask is called when DFJobPhase is "Created".
func genSubTaskStageByTask(logger *zap.Logger, stage metadata.TaskStage) dmpb.SubTask_Stage {
	switch stage {
	case metadata.StageInit:
		return dmpb.SubTask_Creating
	case metadata.StageRunning:
		return dmpb.SubTask_Running
	case metadata.StagePausing:
		return dmpb.SubTask_Pausing
	case metadata.StagePaused:
		return dmpb.SubTask_Paused
	case metadata.StageFinished:
		return dmpb.SubTask_Finished
	case metadata.StageError:
		return dmpb.SubTask_Failed
	case metadata.StageUnscheduled:
		return dmpb.SubTask_Creating
	}
	// this can't happen now
	logger.Warn("task stage is unknown")
	return dmpb.SubTask_Failed
}

func genSubTaskStateByJobState(logger *zap.Logger, state engpb.Job_State) dmpb.SubTask_Stage {
	switch state {
	case engpb.Job_Created, engpb.Job_Running:
		return dmpb.SubTask_Running
	case engpb.Job_Failed:
		return dmpb.SubTask_Failed
	case engpb.Job_Finished:
		return dmpb.SubTask_Finished
	case engpb.Job_Canceled, engpb.Job_Canceling:
		// this can not happen here
		logger.Warn("job state is deleting")
		return dmpb.SubTask_Deleting
	}
	// this can't happen now
	logger.Warn("job state is unknown")
	return dmpb.SubTask_Failed
}

func genSubTaskStateByDMSubTaskStage(logger *zap.Logger, subStatus *openapi.SubTaskStatus) dmpb.SubTask_Stage {
	switch subStatus.Stage {
	case openapi.TaskStageRunning:
		return dmpb.SubTask_Running
	case openapi.TaskStageFinished:
		return dmpb.SubTask_Finished
	case openapi.TaskStageStopped:
		return dmpb.SubTask_Paused
	case openapi.TaskStagePaused:
		if subStatus.ErrorMsg != nil {
			return dmpb.SubTask_Failed
		} else {
			return dmpb.SubTask_Paused
		}
	default:
		// this can not happen here
		logger.Warn("unexpected subtask stage", zap.Any("subTaskStatus", subStatus))
		return dmpb.SubTask_Unknown
	}
}

func jobSource2Dataflow(jobSource entity.DMJobSource) *dmpb.Source {
	source := &dmpb.Source{
		ConnProfileId: jobSource.ConnProfileID.Uint64(),
	}
	// convert route rules
	source.RouteRules = make([]*dmpb.RouteRule, len(jobSource.RouteRules))
	for j, tableRule := range jobSource.RouteRules {
		source.RouteRules[j] = routeRule2Dataflow(&tableRule)
	}

	// convert ba-list
	if jobSource.BaRules != nil {
		source.BaRules = baList2Dataflow((filter.Rules)(*jobSource.BaRules))
	}

	return source
}

func jobSpecEntity2PB(spec *entity.DMJobSpec) *dmpb.JobSpec {
	return &dmpb.JobSpec{
		Id:                uint64(spec.ID),
		DisplayName:       spec.DisplayName,
		CapacityUnit:      spec.CapacityUnit,
		MaxFullPhaseSpeed: int64(spec.MaxFullPhaseSpeed),
		MaxIncrPhaseSpeed: int64(spec.MaxIncrPhaseSpeed),
	}
}

func detail2PB(log *zap.Logger, status *dmpkg.QueryStatusResponse, subTask *dmpb.SubTask, duration time.Duration) error {
	var err error

	if status.Result != nil && len(status.Result.Errors) > 0 {
		log.Warn("some errors in result", zap.Reflect("errors", status.Result.Errors))
		subTask.ProcessError = dmProcessError2PB(status.Result.Errors[0])
		subTask.Stage = dmpb.SubTask_Failed
	}
	switch status.Unit {
	case frameModel.WorkerDMDump:
		subTask.CurrentStep = dmpb.SubTask_Dump
		if len(status.Status) == 0 {
			return nil
		}
		var dumpStatus tiflowdmpb.DumpStatus
		// note: use json.Unmarshal will lead to some error
		// ref from https://github.com/pingcap/tiflow/blob/7dbb20194480d5b437fc3e42c161729c996aac3e/engine/executor/dm/api.go#L38-L40
		if err := jsonpb.Unmarshal(bytes.NewReader(status.Status), &dumpStatus); err != nil {
			return err
		}
		subTask.DumpDetail, err = dumpDetailTiflow2PB(&dumpStatus)
		if err != nil {
			return err
		}
		subTask.DumpDetail.Duration = int64(duration.Seconds())
	case frameModel.WorkerDMLoad:
		subTask.CurrentStep = dmpb.SubTask_Load
		if len(status.Status) == 0 {
			return nil
		}
		var loadStatus tiflowdmpb.LoadStatus
		if err := jsonpb.Unmarshal(bytes.NewReader(status.Status), &loadStatus); err != nil {
			return err
		}
		subTask.LoadDetail, err = loadDetailTiflow2PB(&loadStatus)
		if err != nil {
			return err
		}
		subTask.LoadDetail.Duration = int64(duration.Seconds())
	case frameModel.WorkerDMSync:
		subTask.CurrentStep = dmpb.SubTask_Sync
		if len(status.Status) == 0 {
			return nil
		}
		var syncStatus tiflowdmpb.SyncStatus
		if err := jsonpb.Unmarshal(bytes.NewReader(status.Status), &syncStatus); err != nil {
			return err
		}
		subTask.SyncDetail = syncDetailTiflow2PB(&syncStatus)
		subTask.SyncDetail.Duration = int64(duration.Seconds())
	default:
		// should not happen
		subTask.CurrentStep = dmpb.SubTask_Unsupported
		log.Warn("an unsupported subtask type in dataflow service", zap.Stringer("subtask type", status.Unit))
		return nil
	}
	return nil
}

func dumpDetailEntity2PB(detail *entity.DMDumpStatus) (*dmpb.DumpDetail, error) {
	progress, err := removePercent(detail.Progress)
	if err != nil {
		return nil, err
	}
	return &dmpb.DumpDetail{
		Bps:             detail.Bps,
		Duration:        detail.Duration,
		Progress:        progress,
		TotalTables:     detail.TotalTables,
		CompletedTables: int64(detail.CompletedTables),
		FinishedBytes:   int64(detail.FinishedBytes),
		FinishedRows:    int64(detail.FinishedRows),
	}, nil
}

func loadDetailEntity2PB(detail *entity.DMLoadStatus) (*dmpb.LoadDetail, error) {
	progress, err := removePercent(detail.Progress)
	if err != nil {
		return nil, err
	}
	return &dmpb.LoadDetail{
		FinishedBytes: detail.FinishedBytes,
		TotalBytes:    detail.TotalBytes,
		Bps:           detail.Bps,
		Progress:      progress,
		Duration:      detail.Duration,
	}, nil
}

func loadDetaiImportPB(detail *dataflow.ImportGetResp) *dmpb.LoadDetail {
	currentTime := time.Now()
	startedAt := currentTime
	if detail.CreatedAt != nil {
		startedAt = detail.CreatedAt.AsTime()
	}
	duration := int64(currentTime.Sub(startedAt).Seconds())
	bps := int64(0)
	if duration > 0 {
		bps = int64(float64(detail.ProcessedSourceDataSize) / float64(duration))
	}
	return &dmpb.LoadDetail{
		FinishedBytes: int64(detail.ProcessedSourceDataSize),
		TotalBytes:    int64(detail.TotalSize),
		Bps:           bps,
		Progress:      float64(detail.CompletedPercent+detail.PostImportCompletedPercent) / 2.0,
		Duration:      duration,
	}
}

func dumpDetailTiflow2PB(detail *tiflowdmpb.DumpStatus) (*dmpb.DumpDetail, error) {
	progress, err := removePercent(detail.Progress)
	if err != nil {
		return nil, err
	}
	return &dmpb.DumpDetail{
		Bps:             detail.Bps,
		Progress:        progress,
		TotalTables:     detail.TotalTables,
		CompletedTables: int64(detail.CompletedTables),
		FinishedBytes:   int64(detail.FinishedBytes),
		FinishedRows:    int64(detail.FinishedRows),
	}, nil
}

func loadDetailTiflow2PB(detail *tiflowdmpb.LoadStatus) (*dmpb.LoadDetail, error) {
	progress, err := removePercent(detail.Progress)
	if err != nil {
		return nil, err
	}
	return &dmpb.LoadDetail{
		FinishedBytes: detail.FinishedBytes,
		TotalBytes:    detail.TotalBytes,
		Bps:           detail.Bps,
		Progress:      progress,
	}, nil
}

func syncDetailTiflow2PB(detail *tiflowdmpb.SyncStatus) *dmpb.SyncDetail {
	return &dmpb.SyncDetail{
		Rps:        detail.RecentRps,
		Latency:    detail.SecondsBehindMaster,
		Checkpoint: detail.SyncerBinlog,
	}
}

func normalizeErrMsg(in string, scope string) string {
	switch {
	case strings.Contains(in, "Could not find first log file name in binary log index file"):
		return "The required binary log for migration no longer exists on the source database. Please make sure binary log files are kept for long enough time for migration to succeed."
	case strings.Contains(in, "dial tcp"):
		// connect issues, we need to different between source and target.
		// dial tcp ... i/o timeout
		// dial tcp ... no such host
		// dial tcp ... connect: connection refused
		if scope == terror.ScopeUpstream.String() || strings.Contains(in, "start syncing binlog from remote streamer") {
			// must be source db when trying to get binlog
			return "Failed to connect to the source database using given parameters. Please make sure the source database is up and can be connected using the given parameters."
		} else if scope == terror.ScopeDownstream.String() || strings.Contains(in, "[Lightning:") {
			// must be target db during load
			return "Failed to connect to the TiDB cluster using the given user and password. Please make sure TiDB Cluster is up and can be connected to using the given user and password."
		}
	case strings.Contains(in, "tikv disk full"):
		return "TiDB cluster storage is not enough. Please increase the node storage of TiKV."
	case strings.Contains(in, "Error 9501: Timed-out waiting to acquire database connection"):
		// this error came from AWS RDS proxy
		// https://docs.aws.amazon.com/AmazonRDS/latest/UserGuide/rds-proxy.troubleshooting.html
		return "Failed to connect to the source database. Please check whether the database is available or the maximum connections have been reached."
	}
	return in
}

func dmProcessError2PB(error *dmpkg.ProcessError) *dmpb.ProcessError {
	err := &dmpb.ProcessError{
		ErrCode:    error.ErrCode,
		Message:    error.Message,
		Workaround: error.Workaround,
	}
	if len(err.Message) == 0 {
		err.Message = error.RawCause
	} else if len(error.RawCause) > 0 {
		err.Message += " Raw Cause: " + error.RawCause
	}
	err.Message = normalizeErrMsg(err.Message, error.ErrScope)
	return err
}

// todo: return number type directly from engine
func removePercent(s string) (float64, error) {
	if len(s) == 0 {
		return 0, nil
	}
	if s[len(s)-1] == '%' {
		s = s[:len(s)-1]
	}
	s = strings.ReplaceAll(s, " ", "")

	return strconv.ParseFloat(s, 64)
}