package impl

import (
	"context"
	"fmt"
	"strings"

	"github.com/tidbcloud/dataflow-service/app/event_bus/publisher/center"
	"github.com/tidbcloud/dataflow-service/app/models/biz_changefeed"
	"github.com/tidbcloud/dataflow-service/app/repo/entity"
	"github.com/tidbcloud/eventbus-datastruct/event"
)

type ActionType int

const (
	ActionTypeStart = iota
	ActionTypeEnd
	ActionTypeFailed

	ChangefeedSuffix = "Changefeed"

	ChangefeedStartMsg = "Start to %s changefeed."
	ChangefeedEndMsg   = "Changefeed successfully %s."

	ChangefeedKindMsg = "Task kind: %s."
	ChangefeedNameMsg = "Task name: %s."

	ChangefeedFailedMsg = "Failed to run changefeed."
)

var (
	downstreamType2MsgKindType = map[entity.TiCDCChangefeedDownStreamType]center.MsgKindType{
		entity.TiCDCDownStreamTypeMySQL: center.MysqlChangefeed,
	}
	changefeedStatus2MsgStatusType = map[entity.BizChangefeedOperationStatus]center.MsgStatusType{
		entity.BizChangefeedOperationStatusCreating: center.CreatingStatus,
		entity.BizChangefeedOperationStatusNormal:   center.NormalStatus,
		entity.BizChangefeedOperationStatusFailed:   center.FailedStatus,
		entity.BizChangefeedOperationStatusScaling:  center.ScalingStatus,
		entity.BizChangefeedOperationStatusPaused:   center.StoppedStatus,
		entity.BizChangefeedOperationStatusDeleting: center.DeletingStatus,
		entity.BizChangefeedOperationStatusDeleted:  center.DeletedStatus,
	}
	changefeedStatus2ActionType = map[entity.BizChangefeedOperationStatus]ActionType{
		entity.BizChangefeedOperationStatusCreating: ActionTypeStart,
		entity.BizChangefeedOperationStatusNormal:   ActionTypeEnd,
		entity.BizChangefeedOperationStatusFailed:   ActionTypeFailed,
		entity.BizChangefeedOperationStatusScaling:  ActionTypeStart,
		entity.BizChangefeedOperationStatusPaused:   ActionTypeEnd,
		entity.BizChangefeedOperationStatusDeleting: ActionTypeStart,
		entity.BizChangefeedOperationStatusDeleted:  ActionTypeEnd,
	}
	changefeedStatus2EventStatus = map[entity.BizChangefeedOperationStatus]event.EventStatus{
		entity.BizChangefeedOperationStatusCreating: event.EventStatusStarted,
		entity.BizChangefeedOperationStatusNormal:   event.EventStatusSucceeded,
		entity.BizChangefeedOperationStatusFailed:   event.EventStatusFailed,
		entity.BizChangefeedOperationStatusScaling:  event.EventStatusStarted,
		entity.BizChangefeedOperationStatusPaused:   event.EventStatusSucceeded,
		entity.BizChangefeedOperationStatusDeleting: event.EventStatusStarted,
		entity.BizChangefeedOperationStatusDeleted:  event.EventStatusSucceeded,
	}
)

type bizChangefeedEventBusHooker struct {
	publisher center.EventCenterPublisher
}

func NewBizChangefeedEventBusHooker(publisher center.EventCenterPublisher) biz_changefeed.BizChangefeedHooker {
	return &bizChangefeedEventBusHooker{
		publisher: publisher,
	}
}

func (m *bizChangefeedEventBusHooker) HandleBizChangefeedEvent(
	ctx context.Context,
	event entity.DataflowEventType,
	userID uint64,
	bizChangefeedEntity *entity.BizChangefeed,
) {
	displayEvent := buildChangefeedDisplayEvent(event, userID, bizChangefeedEntity)
	m.publisher.PublishAsync(displayEvent)
}

func buildChangefeedMsg(
	userID uint64, bizChangefeedEntity *entity.BizChangefeed,
) *center.ChangefeedMsg {
	changefeedMsg := &center.ChangefeedMsg{
		BaseMsg: center.BaseMsg{
			Name:         bizChangefeedEntity.Name,
			Type:         downstreamType2MsgKindType[bizChangefeedEntity.DownstreamType],
			Status:       changefeedStatus2MsgStatusType[bizChangefeedEntity.OperationStatus],
			HappenedAtMs: uint64(bizChangefeedEntity.UpdatedAt.UnixMilli()),
		},
		IDMsg: center.IDMsg{
			AuthorID:  userID,
			TenantID:  uint64(bizChangefeedEntity.TenantID),
			ProjectID: uint64(bizChangefeedEntity.ProjectID),
			ClusterID: uint64(bizChangefeedEntity.ClusterID),
		},
		ChangefeedID: uint64(bizChangefeedEntity.ID),
		CdcClusterID: uint64(bizChangefeedEntity.TiCDCClusterID),
	}
	return changefeedMsg
}

func buildChangefeedDisplayEvent(
	event entity.DataflowEventType, userID uint64, bizChangefeedEntity *entity.BizChangefeed,
) *event.DisplayEvent {
	msg := buildChangefeedMsg(userID, bizChangefeedEntity)

	data := center.BuildEventBasicInfos(msg.ChangefeedID, msg.HappenedAtMs, msg.IDMsg)
	data.Detail = msg
	data.DisplayStatus = changefeedStatus2EventStatus[bizChangefeedEntity.OperationStatus]
	data.DisplayEventType = string(event)

	summary := buildChangefeedEventMsg(event, bizChangefeedEntity)
	summary += fmt.Sprintf(ChangefeedKindMsg+" ", msg.Type)
	summary += fmt.Sprintf(ChangefeedNameMsg, msg.Name)
	data.Summary = summary
	return data
}

func buildChangefeedEventMsg(
	event entity.DataflowEventType,
	bizChangefeedEntity *entity.BizChangefeed,
) string {
	actionType := changefeedStatus2ActionType[bizChangefeedEntity.OperationStatus]
	switch actionType {
	case ActionTypeStart:
		return fmt.Sprintf(ChangefeedStartMsg+"\n", changefeedEvent2Action(event))
	case ActionTypeEnd:
		return fmt.Sprintf(ChangefeedEndMsg+"\n", changefeedEvent2Action(event))
	case ActionTypeFailed:
		return ChangefeedFailedMsg + "\n" + fmt.Sprintf("Reason: %s\n", bizChangefeedEntity.ErrMsg)
	default:
		return ""
	}
}

func changefeedEvent2Action(event entity.DataflowEventType) string {
	return strings.TrimSuffix(string(event), ChangefeedSuffix)
}