package tidbtools

import (
	"strconv"
	"time"

	premiumdm "github.com/tidbcloud/dataflow-service-ng/idl/pbgen/proto/premium_dm/v1beta2"
	"github.com/tidbcloud/dataflow-service/idl/pbgen/proto/dm"
)

// DMOverview mirrors `DMOverview` in FE types — the cluster-level DM
// summary surfaced on the Tools overview panel.
type DMOverview struct {
	Version       string `json:"version"`
	VersionSource string `json:"version_source,omitempty"`
}

// DMJob is the FE-facing projection of dm.Job. Only the fields the
// Tools page needs are exposed; IDL-internal fields (e.g. raw JobSpec,
// cluster user, source DB credentials) are intentionally dropped to
// insulate the FE from IDL drift and to avoid leaking secrets.
type DMJob struct {
	ID            string      `json:"id"`
	Name          string      `json:"name"`
	Status        DMJobStatus `json:"status" enums:"running,paused,finished,failed,stopped,unknown"`
	SourceType    string      `json:"source_type,omitempty"`
	TargetCluster string      `json:"target_cluster,omitempty"`
	Mode          DMJobMode   `json:"mode,omitempty" enums:"full,incremental,all"`
	ConnMethod    string      `json:"conn_method,omitempty"`
	CreatedAt     int64       `json:"created_at,omitempty"`
	UpdatedAt     int64       `json:"updated_at,omitempty"`
	ErrorMsg      string      `json:"error_msg,omitempty"`
}

// DMSubTask mirrors `DMSubTask` in FE types.
type DMSubTask struct {
	Name            string      `json:"name"`
	Mode            DMJobMode   `json:"mode" enums:"full,incremental,all"`
	Status          DMJobStatus `json:"status" enums:"running,paused,finished,failed,stopped,unknown"`
	ProgressPercent *int32      `json:"progress_percent,omitempty"`
	BinlogPosition  string      `json:"binlog_position,omitempty"`
	BinlogTs        int64       `json:"binlog_ts,omitempty"`
	RowsPerSec      *int64      `json:"rows_per_sec,omitempty"`
}

// DMJobDetail is the response body for `GET .../dm/jobs/{job_id}`.
// The DMJob fields are inlined into the response payload so the FE
// can consume the same shape as `mockApi.getDMJob`.
type DMJobDetail struct {
	DMJob
	SubTasks []*DMSubTask `json:"sub_tasks"`
}

// ListDMJobsResp is the response body for `GET .../dm/jobs`.
type ListDMJobsResp struct {
	Items []*DMJob `json:"items"`
	Total int64    `json:"total"`
}

// DMPrecheck is the summary row used in the precheck list view.
type DMPrecheck struct {
	ID         string           `json:"id"`
	DMJobName  string           `json:"dm_job_name,omitempty"`
	Status     DMPrecheckStatus `json:"status" enums:"passed,failed,warning,running"`
	StartedAt  int64            `json:"started_at,omitempty"`
	FinishedAt int64            `json:"finished_at,omitempty"`
	Finished   int32            `json:"finished,omitempty"`
	Total      int32            `json:"total,omitempty"`
	ErrMsg     string           `json:"errmsg,omitempty"`
}

// DMPrecheckItem mirrors `DMPrecheckItem` in FE types.
type DMPrecheckItem struct {
	Name   string           `json:"name"`
	Status DMPrecheckStatus `json:"status" enums:"passed,failed,warning,running"`
	Result string           `json:"result,omitempty"`
	ErrMsg string           `json:"errmsg,omitempty"`
}

// DMPrecheckDetail is the response body for `GET .../dm/prechecks/{id}`.
type DMPrecheckDetail struct {
	DMPrecheck
	Items []*DMPrecheckItem `json:"items"`
}

// ListDMPrechecksResp is the response body for `GET .../dm/prechecks`.
type ListDMPrechecksResp struct {
	Items []*DMPrecheck `json:"items"`
	Total int64         `json:"total"`
}

// ----- Production converters --------------------------------------------
// The dataflow-service `dm.Job` IDL is much narrower than the FE shape
// (no top-level status, no source_type, no error_msg). The converters
// below populate what they can and leave the rest at zero values; the
// `omitempty` tags keep the JSON identical to FE expectations.

// ToListDMJobsResp converts a dm.ListJobsResp into the FE-facing DTO.
// Never returns nil; empty responses yield an empty slice so the FE
// never sees `null`.
func ToListDMJobsResp(pb *dm.ListJobsResp) *ListDMJobsResp {
	out := &ListDMJobsResp{Items: make([]*DMJob, 0)}
	if pb == nil {
		return out
	}
	out.Total = int64(pb.Total)
	for _, j := range pb.Jobs {
		if j == nil {
			continue
		}
		out.Items = append(out.Items, toDMJob(j))
	}
	return out
}

// ToGetDMJobResp converts a dm.GetJobResp into the FE-facing detail.
func ToGetDMJobResp(pb *dm.GetJobResp) *DMJobDetail {
	if pb == nil || pb.Job == nil {
		return &DMJobDetail{SubTasks: make([]*DMSubTask, 0)}
	}
	job := toDMJob(pb.Job)
	out := &DMJobDetail{DMJob: *job, SubTasks: make([]*DMSubTask, 0, len(pb.Job.SubTasks))}
	for _, st := range pb.Job.SubTasks {
		if st == nil {
			continue
		}
		out.SubTasks = append(out.SubTasks, &DMSubTask{
			Name:   st.CurrentStep.String(),
			Mode:   DMJobMode(pb.Job.Mode),
			Status: dmSubTaskStageToStatus(st.Stage),
		})
	}
	return out
}

// ToGetDMPrecheckResp converts a dm.GetPrecheckJobResp into the
// FE-facing precheck detail. The narrow upstream RPC carries only the
// item list and aggregate counts; FE-only fields (started_at, dm_job_name)
// are left at zero.
func ToGetDMPrecheckResp(id uint64, pb *dm.GetPrecheckJobResp) *DMPrecheckDetail {
	out := &DMPrecheckDetail{
		DMPrecheck: DMPrecheck{ID: strconv.FormatUint(id, 10)},
		Items:      make([]*DMPrecheckItem, 0),
	}
	if pb == nil {
		return out
	}
	out.Status = normalizePrecheckStatus(pb.Status)
	out.Total = pb.Total
	for _, it := range pb.Items {
		if it == nil {
			continue
		}
		out.Items = append(out.Items, &DMPrecheckItem{
			Name:   it.Desc,
			Status: normalizePrecheckStatus(it.Status),
			// `Reason` may carry source-DB hostnames/usernames — see design
			// doc §9.4. We forward as-is to the FE (which only renders it
			// in a popover) and never log it.
			ErrMsg: it.Reason,
		})
	}
	return out
}

// normalizePrecheckStatus maps the raw string values returned by the DM
// proto (e.g. "success", "warn") to the canonical DMPrecheckStatus values
// used by the FE ("passed", "warning", …).
func normalizePrecheckStatus(s string) DMPrecheckStatus {
	switch s {
	case "success", "passed":
		return DMPrecheckStatusPassed
	case "warn", "warning":
		return DMPrecheckStatusWarning
	case "failed", "error":
		return DMPrecheckStatusFailed
	case "running":
		return DMPrecheckStatusRunning
	default:
		return DMPrecheckStatus(s)
	}
}

// normalizePrecheckStatusFromCounts derives the canonical precheck status by
// also considering the warn/fail item counts. A status reported as "success"
// by the backend is upgraded to "warning" when warnCnt > 0, or to "failed"
// when failedCnt > 0, so the FE always sees an accurate aggregate status.
func normalizePrecheckStatusFromCounts(s string, warnCnt, failedCnt int32) DMPrecheckStatus {
	if failedCnt > 0 {
		return DMPrecheckStatusFailed
	}
	if warnCnt > 0 {
		return DMPrecheckStatusWarning
	}
	return normalizePrecheckStatus(s)
}

// dmStatusPriority drives top-level DM job status aggregation across
// sub-tasks. Higher value wins. Order: Failed > Running > Paused >
// Finished > Unknown (which also covers Creating/Scaling/Deleting).
var dmStatusPriority = map[DMJobStatus]int{
	DMJobStatusFailed:   5,
	DMJobStatusRunning:  4,
	DMJobStatusPaused:   3,
	DMJobStatusFinished: 2,
	DMJobStatusUnknown:  1,
}

func toDMJob(j *dm.Job) *DMJob {
	if j == nil {
		return nil
	}
	out := &DMJob{
		ID:   strconv.FormatUint(j.Id, 10),
		Name: j.Name,
		Mode: DMJobMode(j.Mode),
	}
	bestPriority := 0
	for _, st := range j.SubTasks {
		if st == nil {
			continue
		}
		s := dmSubTaskStageToStatus(st.Stage)
		if p := dmStatusPriority[s]; p > bestPriority {
			bestPriority = p
			out.Status = s
		}
		if out.SourceType == "" && st.Source != nil && st.Source.ConnProfile != nil {
			out.SourceType = st.Source.ConnProfile.Type
			out.ConnMethod = st.Source.ConnProfile.ConnMethod
		}
	}
	if out.Status == "" {
		out.Status = DMJobStatusUnknown
	}
	if j.CreateTime != "" {
		if t, err := time.Parse(time.RFC3339, j.CreateTime); err == nil {
			out.CreatedAt = t.Unix()
		}
	}
	return out
}

// dmSubTaskStageToStatus maps a SubTask_Stage to a FE-facing DMJobStatus.
func dmSubTaskStageToStatus(stage dm.SubTask_Stage) DMJobStatus {
	switch stage {
	case dm.SubTask_Running, dm.SubTask_Creating, dm.SubTask_Deleting, dm.SubTask_Scaling:
		return DMJobStatusRunning
	case dm.SubTask_Pausing, dm.SubTask_Paused:
		return DMJobStatusPaused
	case dm.SubTask_Failed:
		return DMJobStatusFailed
	case dm.SubTask_Finished:
		return DMJobStatusFinished
	default:
		return DMJobStatusUnknown
	}
}

// ToListDMPrechecksResp converts a dm.ListPrechecksResp to the FE-facing
// ListDMPrechecksResp.
func ToListDMPrechecksResp(pb *dm.ListPrechecksResp) *ListDMPrechecksResp {
	out := &ListDMPrechecksResp{Items: make([]*DMPrecheck, 0)}
	if pb == nil {
		return out
	}
	for _, p := range pb.GetPrechecks() {
		if p == nil {
			continue
		}
		item := &DMPrecheck{
			ID:       strconv.FormatUint(p.GetPrecheckId(), 10),
			Status:   normalizePrecheckStatusFromCounts(p.GetStatus(), p.GetWarnCnt(), p.GetFailedCnt()),
			Finished: p.GetSuccessCnt() + p.GetFailedCnt() + p.GetWarnCnt(),
			Total:    p.GetTotal(),
		}
		if ts := p.GetCreatedAt(); ts != nil {
			item.StartedAt = ts.AsTime().Unix()
		}
		out.Items = append(out.Items, item)
	}
	out.Total = int64(pb.GetTotal())
	return out
}

// ----- Premium DM converters (dataflow-service-ng PremiumMigrationService) -----

// ToPremiumListDMJobsResp converts a premium migration list to the FE-facing DTO.
func ToPremiumListDMJobsResp(pb *premiumdm.ListPremiumMigrationsResp) *ListDMJobsResp {
	out := &ListDMJobsResp{Items: make([]*DMJob, 0)}
	if pb == nil {
		return out
	}
	out.Total = int64(pb.TotalSize)
	for _, m := range pb.PremiumMigrations {
		if m == nil {
			continue
		}
		out.Items = append(out.Items, toPremiumDMJob(m))
	}
	return out
}

// ToPremiumGetDMJobResp converts a PremiumMigration to the FE-facing detail.
func ToPremiumGetDMJobResp(pb *premiumdm.PremiumMigration) *DMJobDetail {
	if pb == nil {
		return &DMJobDetail{SubTasks: make([]*DMSubTask, 0)}
	}
	job := toPremiumDMJob(pb)
	out := &DMJobDetail{DMJob: *job, SubTasks: make([]*DMSubTask, 0, len(pb.SubTasks))}
	for _, st := range pb.SubTasks {
		if st == nil {
			continue
		}
		sub := &DMSubTask{
			Status: premiumSubTaskStageToStatus(st.Stage),
			Mode:   job.Mode,
		}
		if st.DumpDetail != nil && st.DumpDetail.Progress > 0 {
			p := int32(st.DumpDetail.Progress)
			sub.ProgressPercent = &p
		} else if st.LoadDetail != nil && st.LoadDetail.Progress > 0 {
			p := int32(st.LoadDetail.Progress)
			sub.ProgressPercent = &p
		}
		if st.SyncDetail != nil {
			sub.BinlogPosition = st.SyncDetail.Checkpoint
			rps := st.SyncDetail.Rps
			sub.RowsPerSec = &rps
		}
		out.SubTasks = append(out.SubTasks, sub)
	}
	return out
}

// ToPremiumListDMPrechecksResp converts a premium precheck list to the FE-facing DTO.
func ToPremiumListDMPrechecksResp(pb *premiumdm.ListPremiumMigrationPrechecksResp) *ListDMPrechecksResp {
	out := &ListDMPrechecksResp{Items: make([]*DMPrecheck, 0)}
	if pb == nil {
		return out
	}
	out.Total = int64(pb.TotalSize)
	for _, p := range pb.Prechecks {
		if p == nil {
			continue
		}
		pc := &DMPrecheck{
			ID:       p.PrecheckId,
			Status:   premiumPrecheckSummaryStatus(p),
			Total:    p.Total,
			Finished: p.SuccessCnt + p.FailedCnt + p.WarnCnt,
		}
		if p.CreatedAt != nil {
			pc.StartedAt = p.CreatedAt.AsTime().Unix()
		}
		out.Items = append(out.Items, pc)
	}
	return out
}

// ToPremiumGetDMPrecheckResp converts a PremiumMigrationPrecheck to the
// FE-facing precheck detail.
func ToPremiumGetDMPrecheckResp(precheckID string, pb *premiumdm.PremiumMigrationPrecheck) *DMPrecheckDetail {
	out := &DMPrecheckDetail{
		DMPrecheck: DMPrecheck{ID: precheckID},
		Items:      make([]*DMPrecheckItem, 0),
	}
	if pb == nil {
		return out
	}
	out.Status = premiumPrecheckStatusToStatus(pb.Status)
	out.Total = pb.Total
	out.Finished = pb.SuccessCnt + pb.FailedCnt + pb.WarnCnt
	for _, it := range pb.Items {
		if it == nil {
			continue
		}
		out.Items = append(out.Items, &DMPrecheckItem{
			Name:   it.Description,
			Status: premiumPrecheckItemStatusToStatus(it.Status),
			ErrMsg: it.Reason,
		})
	}
	return out
}

func toPremiumDMJob(m *premiumdm.PremiumMigration) *DMJob {
	if m == nil {
		return nil
	}
	job := &DMJob{
		ID:     m.MigrationId,
		Name:   m.DisplayName,
		Status: premiumMigrationStateToStatus(m.State),
	}
	switch m.Mode {
	case premiumdm.PremiumDMTaskMode_ALL:
		job.Mode = DMJobModeAll
	case premiumdm.PremiumDMTaskMode_INCREMENTAL:
		job.Mode = DMJobModeIncremental
	}
	if m.CreateTime != nil {
		job.CreatedAt = m.CreateTime.AsTime().Unix()
	}
	return job
}

func premiumMigrationStateToStatus(s premiumdm.PremiumMigration_State) DMJobStatus {
	switch s {
	case premiumdm.PremiumMigration_RUNNING:
		return DMJobStatusRunning
	case premiumdm.PremiumMigration_PAUSED:
		return DMJobStatusPaused
	case premiumdm.PremiumMigration_FAILED:
		return DMJobStatusFailed
	case premiumdm.PremiumMigration_CREATING, premiumdm.PremiumMigration_DELETING:
		return DMJobStatusUnknown
	default:
		return DMJobStatusUnknown
	}
}

func premiumSubTaskStageToStatus(s premiumdm.PremiumDMSubTask_Stage) DMJobStatus {
	switch s {
	case premiumdm.PremiumDMSubTask_RUNNING:
		return DMJobStatusRunning
	case premiumdm.PremiumDMSubTask_PAUSED:
		return DMJobStatusPaused
	case premiumdm.PremiumDMSubTask_FAILED:
		return DMJobStatusFailed
	case premiumdm.PremiumDMSubTask_FINISHED:
		return DMJobStatusFinished
	default:
		return DMJobStatusUnknown
	}
}

// premiumPrecheckSummaryStatus converts a PremiumMigrationPrecheckSummary to
// a DMPrecheckStatus, accounting for partial failures and warnings.
func premiumPrecheckSummaryStatus(p *premiumdm.PremiumMigrationPrecheckSummary) DMPrecheckStatus {
	s := premiumPrecheckStatusToStatus(p.Status)
	// A FINISHED precheck with failures should show as "failed", with only warnings as "warning".
	if s == DMPrecheckStatusPassed && p.Status == premiumdm.PremiumMigrationPrecheck_FINISHED {
		if p.FailedCnt > 0 {
			return DMPrecheckStatusFailed
		}
		if p.WarnCnt > 0 {
			return DMPrecheckStatusWarning
		}
	}
	return s
}

func premiumPrecheckStatusToStatus(s premiumdm.PremiumMigrationPrecheck_Status) DMPrecheckStatus {
	switch s {
	case premiumdm.PremiumMigrationPrecheck_FINISHED:
		return DMPrecheckStatusPassed
	case premiumdm.PremiumMigrationPrecheck_RUNNING, premiumdm.PremiumMigrationPrecheck_PENDING:
		return DMPrecheckStatusRunning
	case premiumdm.PremiumMigrationPrecheck_FAILED, premiumdm.PremiumMigrationPrecheck_CANCELED:
		return DMPrecheckStatusFailed
	default:
		return DMPrecheckStatusRunning
	}
}

func premiumPrecheckItemStatusToStatus(s premiumdm.PremiumDMPrecheckItem_Status) DMPrecheckStatus {
	switch s {
	case premiumdm.PremiumDMPrecheckItem_SUCCESS:
		return DMPrecheckStatusPassed
	case premiumdm.PremiumDMPrecheckItem_WARNING:
		return DMPrecheckStatusWarning
	case premiumdm.PremiumDMPrecheckItem_FAILED:
		return DMPrecheckStatusFailed
	default:
		return DMPrecheckStatusRunning
	}
}