package hook_impl

import (
	"fmt"
	"time"

	"github.com/docker/go-units"
	"github.com/tidbcloud/dataflow-service/app/event_bus/publisher/center"
	"github.com/tidbcloud/dataflow-service/app/repo/entity"
	errhandler "github.com/tidbcloud/dataflow-service/pkg/error-handler"
	"github.com/tidbcloud/dataflow-service/pkg/log"
	"github.com/tidbcloud/eventbus-datastruct/event"
	"gorm.io/gorm"
)

const (
	importStageEncode      = "Encode"
	importStageTiKVImport  = "TiKV-Import"
	importStagePostProcess = "Post-Processing"
)

var importStatus2TypeMap = map[entity.ImportStatus]center.MsgStatusType{
	entity.ImportStatusPreparing: center.PendingStatus,
	entity.ImportStatusImporting: center.RunningStatus,
	entity.ImportStatusCompleted: center.SuccessStatus,
	entity.ImportStatusFailed:    center.FailedStatus,
	entity.ImportStatusCanceled:  center.SkippedStatus,
}

func GetImportHooker() entity.DataflowEventHooker {
	return ih
}

var ih = &ImportHooker{}

type ImportHooker struct{}

func (i *ImportHooker) BeforeSave(tx *gorm.DB, v interface{}) {
	BaseBeforeSave(tx, v, isNewlyCreated, i.checkEntityStatusChanged)
}

func (i *ImportHooker) AfterSave(tx *gorm.DB, v interface{}) {
	BaseAfterSave(tx, v, i.convertToDisplayEvent)
}

func (i *ImportHooker) Type() entity.DataflowEventType {
	return entity.ImportDisplayType
}

func (i *ImportHooker) assertion(tx *gorm.DB, value interface{}) *entity.Import {
	logger := log.GetLogger(tx.Statement.Context)
	imp, ok := value.(*entity.Import)
	if !ok {
		logger.Warn("value is not entity.Import")
		return nil
	}

	return imp
}

func (i *ImportHooker) checkEntityStatusChanged(tx *gorm.DB, value interface{}) bool {
	imp := i.assertion(tx, value)
	if imp == nil {
		return false
	}

	var currentRecord entity.Import
	if err := tx.First(&currentRecord, imp.ID).Error; err != nil {
		return false
	}

	var totalSize, currentProcessSize, newProcessSize uint64
	if currentRecord.ProcessedDataSize != nil {
		currentProcessSize = *currentRecord.ProcessedDataSize
	}
	if imp.ProcessedDataSize != nil {
		newProcessSize = *imp.ProcessedDataSize
	}
	if imp.TotalSize != nil {
		totalSize = *imp.TotalSize
	}
	return currentRecord.Status != imp.Status ||
		currentRecord.TaskPhase != imp.TaskPhase ||
		currentRecord.ComponentVersion != imp.ComponentVersion ||
		(float64(newProcessSize-currentProcessSize) > float64(totalSize)*0.1 &&
			totalSize > units.GiB)
}

func (i *ImportHooker) convertToDisplayEvent(tx *gorm.DB, value interface{}) *event.DisplayEvent {
	imp := i.assertion(tx, value)

	importMsg := i.buildImportMsg(imp)

	data := center.BuildEventBasicInfos(importMsg.ImportId, importMsg.HappenedAtMs, importMsg.IDMsg)
	data.DisplayEventType = string(i.Type())
	data.Detail = importMsg
	data.DisplayStatus = center.MsgStatus2EventMap[importMsg.Status]
	// detail will be not shown to user
	data.Author = imp.Creator
	data.Summary = i.buildImportSummary(imp, importMsg, data.DisplayStatus)

	return data
}

func (i *ImportHooker) buildImportSummary(imp *entity.Import,
	importMsg *center.ImportMsg,
	eventStatus event.EventStatus,
) string {
	switch eventStatus {
	case event.EventStatusStarted:
		return i.getStartedSummary(imp, importMsg)
	case event.EventStatusRunning:
		return i.getRunningSummary(imp, importMsg)
	case event.EventStatusCanceled:
		return fmt.Sprintf("The import task was canceled. Task ID: %d.", importMsg.ImportId)
	case event.EventStatusFailed:
		return fmt.Sprintf("The import task failed. Task ID: %d.", importMsg.ImportId)
	case event.EventStatusSucceeded:
		return fmt.Sprintf("The import task succeeded. Task ID: %d.", importMsg.ImportId)
	default:
		return "Unknown event status."
	}
}

func (i *ImportHooker) getStartedSummary(imp *entity.Import, importMsg *center.ImportMsg) string {
	summary := fmt.Sprintf("The import task started. It is in a preparing state. Task ID: %d", importMsg.ImportId)
	if imp.ComponentVersion != "" {
		summary += fmt.Sprintf(", Lightning Version: %s", imp.ComponentVersion)
	}
	return summary + "."
}

func (i *ImportHooker) getRunningSummary(imp *entity.Import, importMsg *center.ImportMsg) string {
	stage := i.getImportStage(imp)
	summary := fmt.Sprintf("The import task is running. Task ID: %d, Lightning Version: %s",
		importMsg.ImportId, imp.ComponentVersion)
	if stage != "" {
		summary += fmt.Sprintf(", Stage: %s", stage)
	}
	if stage == importStageEncode {
		summary += i.getProcessDataSummary(imp)
	}
	return summary + "."
}

func (i *ImportHooker) getImportStage(imp *entity.Import) string {
	switch {
	case imp.StartedAt != nil && imp.TaskImportingAt == nil:
		return importStageEncode
	case imp.TaskImportingAt != nil && imp.TaskPostproceAt == nil:
		return importStageTiKVImport
	case imp.TaskPostproceAt != nil:
		return importStagePostProcess
	default:
		return ""
	}
}

func (i *ImportHooker) getProcessDataSummary(imp *entity.Import) string {
	if imp.TotalSize != nil && imp.ProcessedDataSize != nil && *imp.TotalSize > 0 {
		return fmt.Sprintf(", Process Data: %s/%s",
			units.BytesSize(float64(*imp.ProcessedDataSize)),
			units.BytesSize(float64(*imp.TotalSize)))
	}
	return ""
}

func (i *ImportHooker) buildImportMsg(imp *entity.Import) *center.ImportMsg {
	importMsg := &center.ImportMsg{
		BaseMsg: center.BaseMsg{
			HappenedAtMs: uint64(imp.UpdatedAt.UnixMilli()),
			Status:       importStatus2TypeMap[imp.Status],
		},
		IDMsg: center.IDMsg{
			// the creator in import is the user email
			ClusterID: uint64(imp.ClusterID),
			ProjectID: uint64(imp.ProjectID),
			TenantID:  uint64(imp.TenantID),
		},
		ImportId:         uint64(imp.ID),
		OriginalError:    imp.Message,
		UserFacingError:  errhandler.PrettyMessage(imp.Message, uint32(imp.DataFormat)),
		ComponentVersion: imp.ComponentVersion,
		TotalTables:      int(imp.TotalTablesCount),
		PendingTables:    int(imp.PendingTables),
		CurrentTables:    len(imp.CurrentTables.Tables),
		CompletedTables:  int(imp.CompletedTables),
	}
	if imp.TotalSize != nil {
		importMsg.TotalSize = int(*imp.TotalSize)
	}
	if imp.ProcessedDataSize != nil {
		importMsg.ProcessSize = int(*imp.ProcessedDataSize)
	}
	if imp.StartedAt != nil {
		importMsg.TaskStartedAt = imp.StartedAt.Format(time.RFC3339)
	}
	if imp.TaskImportingAt != nil {
		importMsg.TaskImportingAt = imp.TaskImportingAt.Format(time.RFC3339)
	}
	if imp.TaskPostproceAt != nil {
		importMsg.TaskPostProcessingAt = imp.TaskPostproceAt.Format(time.RFC3339)
	}
	return importMsg
}