package dm

import (
	"context"
	"fmt"
	"time"

	"github.com/tidbcloud/dataflow-service/app/event_bus/publisher/center"
	"github.com/tidbcloud/dataflow-service/internal/db"
	"github.com/tidbcloud/dataflow-service/internal/db/entity"
	"github.com/tidbcloud/dataflow-service/pkg/log"
	"github.com/tidbcloud/eventbus-datastruct/event"
	"go.uber.org/zap"
	"gorm.io/gorm"
)

const (
	DMDisplayType = "DataMigration"

	DMActionScaleCompleted = "scaleCompleted"
	DMActionTaskPause      = "taskPause"
	DMActionTaskResume     = "taskResume"
)

var dmPhase2TypeMap = map[entity.DFJobPhase]center.MsgStatusType{
	// enum('creating','created','unrecoverable','deleting','deleted','finished','finishing','modifying')
	entity.DFJobPhaseCreating:  center.PendingStatus,
	entity.DFJobPhaseCreated:   center.RunningStatus,
	entity.DFJobPhaseDeleting:  center.DeletingStatus,
	entity.DFJobPhaseDeleted:   center.DeletedStatus,
	entity.DFJobPhaseModifying: center.ScalingStatus,
	entity.DFJobPhaseFinishing: center.RunningStatus,
	entity.DFJobPhaseFinished:  center.SuccessStatus,
}

type DMMsg struct {
	center.BaseMsg
	center.IDMsg

	JobId            uint64
	Mode             entity.DMJobMode
	ImportMode       entity.DMImportMode
	RCU              int
	Stage            entity.DMStageType
	Phase            entity.DFJobPhase
	ComponentVersion string
	OriginalError    string
}

type PublishEventReq struct {
	Job        *Job
	Action     string
	FromSpecID entity.ID
	ToSpecID   entity.ID
}

func PublishEvent(ctx context.Context, req *PublishEventReq) {
	logger := log.GetLogger(ctx)
	event, err := convertToDisplayEvent(ctx, req)
	if err != nil {
		logger.Error("failed to publish dm event", zap.Uint64("jobID", req.Job.DFJob.ID.Uint64()), zap.Error(err))
		return
	}
	center.GetEventCenterPublisher().PublishAsync(event)
}

func convertToDisplayEvent(ctx context.Context, req *PublishEventReq) (*event.DisplayEvent, error) {

	dfJob := req.Job.DFJob
	dmJob := req.Job.DMJob
	logger := log.GetLogger(ctx)
	var fromRCU, toRCU int

	if dmJob == nil {
		var dmJobFromDB entity.DMJob
		if err := db.Transaction(func(tx *gorm.DB) error {
			return entity.NewDMJobQuerySet(tx).IDEq(dfJob.ID).One(&dmJobFromDB)
		}); err != nil {
			logger.Error("failed to get dmJob", zap.Uint64("jobID", dfJob.ID.Uint64()), zap.Error(err))
			return nil, err
		}
		dmJob = &dmJobFromDB
	}

	if dfJob.Phase == entity.DFJobPhaseModifying {
		var dmJobSpecs []entity.DMJobSpec
		if err := db.Transaction(func(tx *gorm.DB) error {
			return entity.NewDMJobSpecQuerySet(tx).IDIn(req.FromSpecID, req.ToSpecID).All(&dmJobSpecs)
		}); err != nil {
			logger.Error("failed to get dmJob spec",
				zap.Uint64("jobID", dfJob.ID.Uint64()),
				zap.Uint64("fromSpecID", req.FromSpecID.Uint64()),
				zap.Uint64("toSepcID", req.ToSpecID.Uint64()),
				zap.Error(err))
			return nil, err
		}
		for _, jobSpec := range dmJobSpecs {
			switch jobSpec.ID {
			case req.FromSpecID:
				fromRCU = int(jobSpec.CapacityUnit)
			case req.ToSpecID:
				toRCU = int(jobSpec.CapacityUnit)
			}
		}
	} else {
		var dmJobSpec entity.DMJobSpec
		if err := db.Transaction(func(tx *gorm.DB) error {
			return entity.NewDMJobSpecQuerySet(tx).IDEq(dmJob.SpecID).One(&dmJobSpec)
		}); err != nil {
			logger.Error("failed to get dmJob spec", zap.Uint64("jobID", dfJob.ID.Uint64()), zap.Error(err))
			return nil, err
		}
		fromRCU = int(dmJobSpec.CapacityUnit)
		toRCU = int(dmJobSpec.CapacityUnit)
	}

	var dmJobSource entity.DMJobSource
	if err := db.Transaction(func(tx *gorm.DB) error {
		return entity.NewDMJobSourceQuerySet(tx).JobIDEq(dmJob.ID).One(&dmJobSource)
	}); err != nil {
		logger.Error("failed to get dmJob source", zap.Uint64("jobID", dmJob.ID.Uint64()), zap.Error(err))
		return nil, err
	}

	stage := GetDMStage(dmJob, &dmJobSource)

	dmMsg := &DMMsg{
		BaseMsg: center.BaseMsg{
			HappenedAtMs: uint64(time.Now().UnixMilli()),
			Status:       dmPhase2TypeMap[dfJob.Phase],
		},
		IDMsg: center.IDMsg{
			// the creator in dmJobort is the user email
			AuthorID:  dfJob.CreatorUID.Uint64(),
			ClusterID: uint64(dmJob.ClusterID),
			ProjectID: uint64(dmJob.ProjectID),
			TenantID:  uint64(dmJob.TenantID),
		},
		JobId:            uint64(dmJob.ID),
		OriginalError:    dfJob.ErrorMessage,
		ComponentVersion: dmJob.ComponentVersion,
		Mode:             dmJob.Mode,
		ImportMode:       dmJob.ImportMode,
		RCU:              toRCU,
		Stage:            stage,
		Phase:            dfJob.Phase,
	}

	data := center.BuildEventBasicInfos(dmMsg.JobId, dmMsg.HappenedAtMs, dmMsg.IDMsg)
	data.DisplayEventType = DMDisplayType
	data.Detail = dmMsg
	data.DisplayStatus = getDisplayStatus(ctx, dfJob)

	var summary string
	switch dfJob.Phase {
	case entity.DFJobPhaseCreating:
		summary = fmt.Sprintf("The dm task started. It is in a preparing state. "+
			"Task ID: %d, Mode: %s, Import Mode: %s, RCU: %d.",
			dmJob.ID, dmJob.Mode, dmJob.ImportMode, toRCU)
	case entity.DFJobPhaseCreated:
		switch req.Action {
		case DMActionScaleCompleted:
			summary = fmt.Sprintf("The dm task modification completed. Task ID: %d, Current Stage: %s, RCU: %d, Component Version: %s.",
				dmJob.ID, stage, toRCU, dmJob.ComponentVersion)
		case DMActionTaskPause:
			dmMsg.Status = center.StoppedStatus
			data.DisplayStatus = event.EventStatusPaused
			summary = fmt.Sprintf("The dm task paused. Task ID: %d, Current Stage: %s, RCU: %d, Component Version: %s.",
				dmJob.ID, stage, toRCU, dmJob.ComponentVersion)
		case DMActionTaskResume:
			data.DisplayStatus = event.EventStatusResumed
			summary = fmt.Sprintf("The dm task resumed. Task ID: %d, Current Stage: %s, RCU: %d, Component Version: %s.",
				dmJob.ID, stage, toRCU, dmJob.ComponentVersion)
		default:
			summary = fmt.Sprintf("The dm task is running. Task ID: %d, Current Stage: %s, RCU: %d, Component Version: %s.",
				dmJob.ID, stage, toRCU, dmJob.ComponentVersion)
		}
	case entity.DFJobPhaseModifying:
		summary = fmt.Sprintf("The dm task is being modified. Task ID: %d, Component Version: %s", dmJob.ID, dmJob.ComponentVersion)
		if fromRCU != toRCU {
			summary += fmt.Sprintf(", scaling RCU from %d to %d",
				fromRCU, toRCU)
		}
		summary += "."
	case entity.DFJobPhaseFinishing:
		summary = fmt.Sprintf("The dm task is being finished. Task ID: %d.", dmJob.ID)
	case entity.DFJobPhaseFinished:
		summary = fmt.Sprintf("The dm task succeeded. Task ID: %d.", dmJob.ID)
	case entity.DFJobPhaseDeleting:
		summary = fmt.Sprintf("The dm task is being deleted. Task ID: %d.", dmJob.ID)
	case entity.DFJobPhaseDeleted:
		summary = fmt.Sprintf("The dm task was deleted. Task ID: %d.", dmJob.ID)
	}

	data.Summary = summary

	return data, nil
}

func getDisplayStatus(ctx context.Context, dfJob *entity.DFJob) event.EventStatus {

	switch dfJob.Phase {
	case entity.DFJobPhaseCreating:
		return event.EventStatusStarted
	case entity.DFJobPhaseCreated:
		return event.EventStatusRunning
	case entity.DFJobPhaseDeleting:
		return event.EventStatusRunning
	case entity.DFJobPhaseDeleted:
		if dfJob.ErrorMessage == "" {
			return event.EventStatusSucceeded
		}
		return event.EventStatusFailed
	case entity.DFJobPhaseModifying:
		return event.EventStatusRunning
	case entity.DFJobPhaseFinishing:
		return event.EventStatusRunning
	case entity.DFJobPhaseFinished:
		return event.EventStatusSucceeded
	default:
		logger := log.GetLogger(ctx)
		logger.Warn("unknown dm job phase", zap.Uint64("jobID", dfJob.ID.Uint64()), zap.String("phase", string(dfJob.Phase)))
		return event.EventStatusFailed
	}
}