package pimport

import (
	"context"
	"fmt"
	"time"

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

// PublishImportIntoEvent publishes a DisplayEvent for an Import Into task status transition.
func PublishImportIntoEvent(
	ctx context.Context,
	publisher center.EventCenterPublisher,
	task *entity.ImportIntoTask,
) {
	if publisher == nil || task == nil {
		return
	}
	ev := convertImportIntoToDisplayEvent(task)
	if ev == nil {
		return
	}
	publisher.PublishAsync(ev)
	log.GetLogger(ctx).Info("published import into display event",
		zap.Uint64("importID", task.ID.Uint64()),
		zap.String("status", string(task.Status)),
		zap.String("displayStatus", string(ev.DisplayStatus)))
}

func publishImportIntoEventIfStatusChanged(
	ctx context.Context,
	publisher center.EventCenterPublisher,
	task *entity.ImportIntoTask,
	previousStatus entity.ImportIntoStatus,
) {
	if task == nil || previousStatus == task.Status {
		return
	}
	PublishImportIntoEvent(ctx, publisher, task)
}

func convertImportIntoToDisplayEvent(task *entity.ImportIntoTask) *event.DisplayEvent {
	msgStatus, displayStatus, summary := importIntoStatusPresentation(task)
	if summary == "" {
		return nil
	}
	msg := &center.ImportIntoMsg{
		BaseMsg: center.BaseMsg{
			Name:         task.Name,
			Status:       msgStatus,
			HappenedAtMs: uint64(time.Now().UnixMilli()),
		},
		IDMsg: center.IDMsg{
			AuthorID:  task.CreatorID,
			TenantID:  task.TenantID.Uint64(),
			ProjectID: task.ProjectID.Uint64(),
			ClusterID: task.ClusterID.Uint64(),
		},
		ImportId:      task.ID.Uint64(),
		OriginalError: task.Message,
		TotalSize:     task.TotalSize,
	}
	data := center.BuildEventBasicInfos(msg.ImportId, msg.HappenedAtMs, msg.IDMsg)
	data.Author = task.Creator
	data.DisplayEventType = string(entity.ImportDisplayType)
	data.DisplayStatus = displayStatus
	data.Detail = msg
	data.Summary = summary
	return data
}

func importIntoStatusPresentation(
	task *entity.ImportIntoTask,
) (center.MsgStatusType, event.EventStatus, string) {
	switch task.Status {
	case entity.ImportIntoStatusPreparing:
		return center.PendingStatus, event.EventStatusStarted, fmt.Sprintf(
			"The import task started. Task ID: %d, Name: %s.",
			task.ID.Uint64(), task.Name)
	case entity.ImportIntoStatusImporting, entity.ImportIntoStatusScalingUp, entity.ImportIntoStatusCanceling:
		// These are internal state-machine transitions. The activity feed only
		// reports the accepted operation and its natural terminal outcome.
		return "", "", ""
	case entity.ImportIntoStatusCompleted:
		return center.SuccessStatus, event.EventStatusSucceeded, fmt.Sprintf(
			"The import task was succeeded. Task ID: %d, Name: %s.",
			task.ID.Uint64(), task.Name)
	case entity.ImportIntoStatusFailed:
		return center.FailedStatus, event.EventStatusFailed, fmt.Sprintf(
			"The import task was failed. Task ID: %d, Name: %s.",
			task.ID.Uint64(), task.Name)
	case entity.ImportIntoStatusCanceled:
		return center.TerminatedStatus, event.EventStatusCanceled, fmt.Sprintf(
			"The import task was canceled. Task ID: %d, Name: %s.",
			task.ID.Uint64(), task.Name)
	default:
		return "", "", ""
	}
}