package clustereventconsumer

import (
	"context"
	"errors"
	"fmt"
	"slices"
	"strconv"
	"strings"

	"github.com/samber/lo"
	"github.com/tidbcloud/eventbus-datastruct/event"
	"github.com/tidbcloud/tiinf/monitor/tilog"
	"github.com/tidbcloud/tiinf/transport/tieventbus"
	"go.uber.org/zap"
	"gorm.io/gorm"

	globaldedicatedv1 "github.com/tidbcloud/tidb-management-service/api/spec/global/dedicated/v1"
	globalngv1 "github.com/tidbcloud/tidb-management-service/api/spec/global/nextgen/v1"
	openapidedicatedv1beta1 "github.com/tidbcloud/tidb-management-service/api/spec/tidb_cloud_open_api/dedicated/v1beta1"
	"github.com/tidbcloud/tidb-management-service/tidb-mgmt-service/domain/entity"
	eventutil "github.com/tidbcloud/tidb-management-service/utilities/clusterevent"
	"github.com/tidbcloud/tidb-management-service/utilities/enum"
	"github.com/tidbcloud/tidb-management-service/utilities/spu/storagetype"
)

var (
	renderTidbTopologyStr = func(count uint32, spuName string) string {
		return fmt.Sprintf(`TiDB x%d (%s)`, count, spuName)
	}
	renderTiproxyTopologyStr = func(count uint32, spuName string) string {
		return fmt.Sprintf(`TiProxy x%d (%s)`, count, spuName)
	}
	renderTikvTopologyStr = func(count, sizeGi uint32, spuName, storageType string, dataIops, raftIops, dataAdditionalIops *uint32) string {
		spec := fmt.Sprintf("%s,%dGiB,%s", spuName, sizeGi, storageType)
		if storageType == openapidedicatedv1beta1.Cluster_StorageNodeSetting_Plus.String() && dataIops != nil {
			spec += fmt.Sprintf(",%dPlusDataDiskIOPS", *dataIops)
		}
		if raftIops != nil {
			spec += fmt.Sprintf(",%dRaftIOPS", *raftIops)
		}
		if dataAdditionalIops != nil {
			spec += fmt.Sprintf(",%dDataAdditionalIOPS", *dataAdditionalIops)
		}
		return fmt.Sprintf(`TiKV x%d (%s)`, count, spec)
	}
	renderTiflashTopologyStr = func(count, sizeGi uint32, spuName, storageType string) string {
		return fmt.Sprintf(`TiFlash x%d (%s,%dGiB,%s)`, count, spuName, sizeGi, storageType)
	}
)

var (
	clusterEventSummaryMap = map[eventutil.EventType]string{
		eventutil.EventTypeCreating:               "Start to create cluster.",
		eventutil.EventTypeCreated:                "Cluster successfully created.",
		eventutil.EventTypePausing:                "Start to pause cluster.",
		eventutil.EventTypePaused:                 "Cluster successfully paused.",
		eventutil.EventTypeResuming:               "Start to resume cluster.",
		eventutil.EventTypeResumed:                "Cluster successfully resumed.",
		eventutil.EventTypeRestoring:              "Start to restore cluster.",
		eventutil.EventTypeRestored:               "Cluster successfully restored.",
		eventutil.EventTypeScaling:                "Start to modify cluster size.",
		eventutil.EventTypeScaled:                 "Cluster size successfully modified.",
		eventutil.EventTypeDeleting:               "Start to delete Cluster.",
		eventutil.EventTypeDeleted:                "Cluster successfully deleted.",
		eventutil.EventTypeApplying:               "Start to apply cluster configuration or version.",
		eventutil.EventTypeApplied:                "Cluster configuration or version successfully applied.",
		eventutil.EventTypeScheduledResume:        "Normal Pause Time is over, cluster will be resumed automatically tomorrow.",
		eventutil.EventTypeUpdatingPublicEndpoint: "Start to update public endpoint.",
		eventutil.EventTypeUpdatedPublicEndpoint:  "Public endpoint successfully updated.",

		eventutil.EventTypeUpdatingCapacity: "Start to update cluster capacity.",
		eventutil.EventTypeUpdatedCapacity:  "Cluster capacity successfully updated.",

		eventutil.EventTypeUpdatingAuditLogConfig: "Start to update cluster audit log config.",
		eventutil.EventTypeUpdatedAuditLogConfig:  "Cluster Audit log config successfully updated.",
		eventutil.EventTypeCreatingPrivateLink:    "Start to create private link endpoint.",
		eventutil.EventTypeCreatedPrivateLink:     "Private link endpoint successfully created.",

		eventutil.EventTypeInsufficientResource: "Current limitations in resource allocation from the cloud provider are preventing cluster operations.",

		eventutil.EventTypeShutdownUpgrade:  "Start to shutdown upgrade cluster.",
		eventutil.EventTypeShutdownUpgraded: "Cluster successfully shutdown upgraded.",

		eventutil.EventTypeProjectChanged:      "Cluster successfully moved to a different project.",
		eventutil.EventTypeRootPasswordRotated: "Cluster root password successfully rotated.",
	}

	clusterEventTypeMap = map[eventutil.EventType]string{
		eventutil.EventTypeCreating:        "CreateCluster",
		eventutil.EventTypeCreated:         "CreateCluster",
		eventutil.EventTypePausing:         "PauseCluster",
		eventutil.EventTypePaused:          "PauseCluster",
		eventutil.EventTypeResuming:        "ResumeCluster",
		eventutil.EventTypeResumed:         "ResumeCluster",
		eventutil.EventTypeRestoring:       "RestoreCluster",
		eventutil.EventTypeRestored:        "RestoreCluster",
		eventutil.EventTypeScaling:         "ModifyClusterSize",
		eventutil.EventTypeScaled:          "ModifyClusterSize",
		eventutil.EventTypeDeleting:        "DeleteCluster",
		eventutil.EventTypeDeleted:         "DeleteCluster",
		eventutil.EventTypeApplying:        "ApplyClusterConfig",
		eventutil.EventTypeApplied:         "ApplyClusterConfig",
		eventutil.EventTypeScheduledResume: "ScheduledResumeCluster",

		eventutil.EventTypeUpdatingPublicEndpoint: "UpdatePublicEndpoint",
		eventutil.EventTypeUpdatedPublicEndpoint:  "UpdatePublicEndpoint",

		eventutil.EventTypeUpdatingCapacity: "UpdateClusterCapacity",
		eventutil.EventTypeUpdatedCapacity:  "UpdateClusterCapacity",

		eventutil.EventTypeUpdatingAuditLogConfig: "UpdateAuditLogConfig",
		eventutil.EventTypeUpdatedAuditLogConfig:  "UpdateAuditLogConfig",
		eventutil.EventTypeCreatingPrivateLink:    "CreatePrivateLink",
		eventutil.EventTypeCreatedPrivateLink:     "CreatePrivateLink",

		eventutil.EventTypeInsufficientResource: "InsufficientVMResource",

		eventutil.EventTypeShutdownUpgrade:  "ShutdownUpgradeCluster",
		eventutil.EventTypeShutdownUpgraded: "ShutdownUpgradeClusterSucceeded",

		eventutil.EventTypeProjectChanged:      "MoveClusterProject",
		eventutil.EventTypeRootPasswordRotated: "RotateRootPassword",
	}

	clusterEventStatusMap = map[eventutil.EventType]event.EventStatus{
		eventutil.EventTypeCreating:        event.EventStatusStarted,
		eventutil.EventTypeCreated:         event.EventStatusSucceeded,
		eventutil.EventTypePausing:         event.EventStatusStarted,
		eventutil.EventTypePaused:          event.EventStatusSucceeded,
		eventutil.EventTypeResuming:        event.EventStatusStarted,
		eventutil.EventTypeResumed:         event.EventStatusSucceeded,
		eventutil.EventTypeRestoring:       event.EventStatusStarted,
		eventutil.EventTypeRestored:        event.EventStatusSucceeded,
		eventutil.EventTypeScaling:         event.EventStatusStarted,
		eventutil.EventTypeScaled:          event.EventStatusSucceeded,
		eventutil.EventTypeDeleting:        event.EventStatusStarted,
		eventutil.EventTypeDeleted:         event.EventStatusSucceeded,
		eventutil.EventTypeApplying:        event.EventStatusStarted,
		eventutil.EventTypeApplied:         event.EventStatusSucceeded,
		eventutil.EventTypeScheduledResume: event.EventStatusSucceeded,

		eventutil.EventTypeUpdatingPublicEndpoint: event.EventStatusStarted,
		eventutil.EventTypeUpdatedPublicEndpoint:  event.EventStatusSucceeded,
		eventutil.EventTypeUpdatingCapacity:       event.EventStatusStarted,
		eventutil.EventTypeUpdatedCapacity:        event.EventStatusSucceeded,

		eventutil.EventTypeUpdatingAuditLogConfig: event.EventStatusStarted,
		eventutil.EventTypeUpdatedAuditLogConfig:  event.EventStatusSucceeded,
		eventutil.EventTypeCreatingPrivateLink:    event.EventStatusStarted,
		eventutil.EventTypeCreatedPrivateLink:     event.EventStatusSucceeded,

		eventutil.EventTypeInsufficientResource: event.EventStatusRunning,

		eventutil.EventTypeShutdownUpgrade:  event.EventStatusStarted,
		eventutil.EventTypeShutdownUpgraded: event.EventStatusSucceeded,

		eventutil.EventTypeProjectChanged:      event.EventStatusSucceeded,
		eventutil.EventTypeRootPasswordRotated: event.EventStatusSucceeded,
	}

	maintenanceWindowEventSummaryMap = map[eventutil.MaintenanceWindowEventType]string{
		eventutil.MaintenanceWindowEventTypeExecutionRunning:  "Cluster maintenance task started.",
		eventutil.MaintenanceWindowEventTypeExecutionDone:     "Cluster maintenance task succeeded.",
		eventutil.MaintenanceWindowEventTypeExecutionCanceled: "Cluster maintenance task canceled.",
	}

	maintenanceWindowEventTypeMap = map[eventutil.MaintenanceWindowEventType]string{
		eventutil.MaintenanceWindowEventTypeExecutionRunning:  "MaintenanceTask",
		eventutil.MaintenanceWindowEventTypeExecutionDone:     "MaintenanceTask",
		eventutil.MaintenanceWindowEventTypeExecutionCanceled: "MaintenanceTask",
	}

	maintenanceWindowEventStatusMap = map[eventutil.MaintenanceWindowEventType]event.EventStatus{
		eventutil.MaintenanceWindowEventTypeExecutionRunning:  event.EventStatusStarted,
		eventutil.MaintenanceWindowEventTypeExecutionDone:     event.EventStatusSucceeded,
		eventutil.MaintenanceWindowEventTypeExecutionCanceled: event.EventStatusCanceled,
	}
)

type eventDisplayMeta struct {
	Summary          string
	DisplayEventType string
	Status           event.EventStatus
}

var (
	resourcePoolEventMetaMap = map[eventutil.EventType]eventDisplayMeta{
		eventutil.EventTypeCreating: {
			Summary:          "Start to create resource pool.",
			DisplayEventType: "CreateResourcePool",
			Status:           event.EventStatusStarted,
		},
		eventutil.EventTypeCreated: {
			Summary:          "Resource pool successfully created.",
			DisplayEventType: "CreateResourcePool",
			Status:           event.EventStatusSucceeded,
		},
		eventutil.EventTypeDeleting: {
			Summary:          "Start to delete resource pool.",
			DisplayEventType: "DeleteResourcePool",
			Status:           event.EventStatusStarted,
		},
		eventutil.EventTypeDeleted: {
			Summary:          "Resource pool successfully deleted.",
			DisplayEventType: "DeleteResourcePool",
			Status:           event.EventStatusSucceeded,
		},
		eventutil.EventTypePausing: {
			Summary:          "Start to pause resource pool.",
			DisplayEventType: "PauseResourcePool",
			Status:           event.EventStatusStarted,
		},
		eventutil.EventTypePaused: {
			Summary:          "Resource pool successfully paused.",
			DisplayEventType: "PauseResourcePool",
			Status:           event.EventStatusSucceeded,
		},
		eventutil.EventTypeResuming: {
			Summary:          "Start to resume resource pool.",
			DisplayEventType: "ResumeResourcePool",
			Status:           event.EventStatusStarted,
		},
		eventutil.EventTypeResumed: {
			Summary:          "Resource pool successfully resumed.",
			DisplayEventType: "ResumeResourcePool",
			Status:           event.EventStatusSucceeded,
		},
		eventutil.EventTypeResourcePoolDefaultSet: {
			Summary:          "Successfully set the default resource pool.",
			DisplayEventType: "SetDefaultResourcePool",
			Status:           event.EventStatusSucceeded,
		},
		eventutil.EventTypeResourcePoolInstanceLimitUpdated: {
			Summary:          "Successfully updated instance limit.",
			DisplayEventType: "UpdateResourcePoolInstanceLimit",
			Status:           event.EventStatusSucceeded,
		},
		eventutil.EventTypeInstanceCreatingOnResourcePool: {
			Summary:          "Start to create instance on resource pool.",
			DisplayEventType: "ResourcePoolCreateInstance",
			Status:           event.EventStatusStarted,
		},
		eventutil.EventTypeInstanceCreatedOnResourcePool: {
			Summary:          "Instance successfully created on resource pool.",
			DisplayEventType: "ResourcePoolCreateInstance",
			Status:           event.EventStatusSucceeded,
		},
		eventutil.EventTypeInstanceDeletingOnResourcePool: {
			Summary:          "Start to delete instance from resource pool.",
			DisplayEventType: "ResourcePoolDeleteInstance",
			Status:           event.EventStatusStarted,
		},
		eventutil.EventTypeInstanceDeletedOnResourcePool: {
			Summary:          "Instance successfully deleted from resource pool.",
			DisplayEventType: "ResourcePoolDeleteInstance",
			Status:           event.EventStatusSucceeded,
		},
	}

	instanceEventMetaMap = map[eventutil.EventType]eventDisplayMeta{
		eventutil.EventTypeCreating: {
			Summary:          "Start to create instance.",
			DisplayEventType: "CreateInstance",
			Status:           event.EventStatusStarted,
		},
		eventutil.EventTypeCreated: {
			Summary:          "Instance successfully created.",
			DisplayEventType: "CreateInstance",
			Status:           event.EventStatusSucceeded,
		},
		eventutil.EventTypeApplied: {
			Summary:          "Instance configuration or version successfully applied.",
			DisplayEventType: "ApplyClusterConfig",
			Status:           event.EventStatusSucceeded,
		},
		eventutil.EventTypeUpdatedCapacity: {
			Summary:          "Instance capacity successfully updated.",
			DisplayEventType: "UpdateClusterCapacity",
			Status:           event.EventStatusSucceeded,
		},
		eventutil.EventTypeRestoring: {
			Summary:          "Start to restore instance.",
			DisplayEventType: "RestoreInstance",
			Status:           event.EventStatusStarted,
		},
		eventutil.EventTypeRestored: {
			Summary:          "Instance successfully restored.",
			DisplayEventType: "RestoreInstance",
			Status:           event.EventStatusSucceeded,
		},
		eventutil.EventTypeDeleting: {
			Summary:          "Start to delete instance.",
			DisplayEventType: "DeleteInstance",
			Status:           event.EventStatusStarted,
		},
		eventutil.EventTypeDeleted: {
			Summary:          "Instance successfully deleted.",
			DisplayEventType: "DeleteInstance",
			Status:           event.EventStatusSucceeded,
		},
		eventutil.EventTypeUpdatingPublicEndpoint: {
			Summary:          "Start to update public endpoint.",
			DisplayEventType: "UpdatePublicEndpoint",
			Status:           event.EventStatusStarted,
		},
		eventutil.EventTypeUpdatedPublicEndpoint: {
			Summary:          "Public endpoint successfully updated.",
			DisplayEventType: "UpdatePublicEndpoint",
			Status:           event.EventStatusSucceeded,
		},
		eventutil.EventTypeUpdatingCapacity: {
			Summary:          "Start to update cluster capacity.",
			DisplayEventType: "UpdateClusterCapacity",
			Status:           event.EventStatusStarted,
		},
		eventutil.EventTypeCreatingPrivateLink: {
			Summary:          "Start to create private link endpoint.",
			DisplayEventType: "CreatePrivateLink",
			Status:           event.EventStatusStarted,
		},
		eventutil.EventTypeCreatedPrivateLink: {
			Summary:          "Private link endpoint successfully created.",
			DisplayEventType: "CreatePrivateLink",
			Status:           event.EventStatusSucceeded,
		},
	}

	resourcePoolEventSummaryMap map[eventutil.EventType]string
	resourcePoolEventTypeMap    map[eventutil.EventType]string
	resourcePoolEventStatusMap  map[eventutil.EventType]event.EventStatus

	instanceEventSummaryMap map[eventutil.EventType]string
	instanceEventTypeMap    map[eventutil.EventType]string
	instanceEventStatusMap  map[eventutil.EventType]event.EventStatus
)

func init() {
	initResourcePoolEventMaps()
	initInstanceEventMaps()
}

func initResourcePoolEventMaps() {
	resourcePoolEventSummaryMap = make(map[eventutil.EventType]string, len(resourcePoolEventMetaMap))
	resourcePoolEventTypeMap = make(map[eventutil.EventType]string, len(resourcePoolEventMetaMap))
	resourcePoolEventStatusMap = make(map[eventutil.EventType]event.EventStatus, len(resourcePoolEventMetaMap))

	for eventType, meta := range resourcePoolEventMetaMap {
		resourcePoolEventSummaryMap[eventType] = meta.Summary
		resourcePoolEventTypeMap[eventType] = meta.DisplayEventType
		resourcePoolEventStatusMap[eventType] = meta.Status
	}
}

func initInstanceEventMaps() {
	instanceEventSummaryMap = make(map[eventutil.EventType]string, len(instanceEventMetaMap))
	instanceEventTypeMap = make(map[eventutil.EventType]string, len(instanceEventMetaMap))
	instanceEventStatusMap = make(map[eventutil.EventType]event.EventStatus, len(instanceEventMetaMap))

	for eventType, meta := range instanceEventMetaMap {
		instanceEventSummaryMap[eventType] = meta.Summary
		instanceEventTypeMap[eventType] = meta.DisplayEventType
		instanceEventStatusMap[eventType] = meta.Status
	}
}

// Please refer to display event development doc https://pingcap.feishu.cn/wiki/wikcn1hAL93rIBbibYF41eTgL7c.
func (c *clusterEventConsumerManagerImp) sendToDisplayEvent(ctx context.Context, ev *tieventbus.Event) error {
	logger := tilog.WithContextS(ctx)

	eventBody, ok := ev.Data.(*eventutil.ClusterEvent)
	if !ok {
		logger.Errorf("event id %s is not a ClusterEvent", ev.ID)
		return nil
	}

	if eventBody.ClusterType == eventutil.ClusterTypeTidbxResourcePool {
		return c.sendResourcePoolToDisplayEvent(ctx, ev)
	}

	if eventBody.ClusterType == eventutil.ClusterTypeTidbxInstance && isNextGenInstancePlan(eventBody.ServicePlan) {
		return c.sendInstanceToDisplayEvent(ctx, ev)
	}

	logger = logger.With(zap.Uint64("clusterID", eventBody.ClusterID))
	logger.Infof(
		"cluster display event consumer, try to handle event, id=%s, clusterID=%d, eventType=%s",
		ev.ID,
		eventBody.ClusterID,
		eventBody.EventType,
	)

	// Filter out events that are not in the whitelist.
	displayMeta, exist := displayMetaForClusterEvent(eventBody)
	if !exist {
		return nil
	}
	summary := displayMeta.Summary

	// Filter out events that not belonged to an existing cluster.
	cluster, err := c.clusterRepo.GetCluster(ctx, eventBody.ClusterID, true)
	if err != nil {
		if errors.Is(err, gorm.ErrRecordNotFound) {
			logger.Infof("ignore cluster event: cluster %d not found in mgmt service", eventBody.ClusterID)
			return nil
		}
		logger.Error("failed to get cluster", zap.Error(err))
		return err
	}

	// Enrich summary with cluster information.
	switch eventBody.EventType {
	case eventutil.EventTypeCreated:
		if eventBody.IsDevTier {
			break
		}

		globalCluster, err := c.dedicatedCli.GetCluster(
			ctx,
			&globaldedicatedv1.GetClusterRequest{ClusterId: eventBody.ClusterID},
		)
		if err != nil {
			logger.Error("failed to get cluster from dedicated global", zap.Error(err))
		}
		summary += fmt.Sprintf(
			`
Version: %s.
Topology: %s.`,
			globalCluster.GetVersion(),
			createTopologyToString(globalCluster),
		)

	case eventutil.EventTypeScaling:
		if isModifyClusterVolumeEvent(eventBody) {
			break
		}

		if eventBody.ServicePlan == "Premium_Reserved" {
			templateTopology := c.loadTopologyFromReservedCapacityTemplate(ctx, cluster, eventBody)
			if templateTopology != nil {
				eventBody.ExtraInfo.TopologyBefore = templateTopology
				eventBody.ExtraInfo.TopologyAfter = templateTopology
			}
		}

		getTIdbGroupName := func(tidbNodeGroup *eventutil.TidbNodeGroup) (string, error) {
			if tidbNodeGroup.DisplayName != "" {
				return tidbNodeGroup.DisplayName, nil
			}

			resp, err := c.dedicatedCli.ListTidbNodeGroups(ctx, &globaldedicatedv1.ListTidbNodeGroupsRequest{
				ClusterIds: []uint64{eventBody.ClusterID},
				Ids:        []uint64{tidbNodeGroup.ID},
			})
			if err != nil {
				return "", err
			}

			if len(resp.TidbNodeGroups) == 0 {
				return "", nil
			}

			return resp.TidbNodeGroups[0].DisplayName, nil
		}

		summary += fmt.Sprintf(
			`
Topology before: %s.
Topology after: %s.`,
			scaleTopologyToString(ctx, eventBody.ExtraInfo.TopologyBefore, getTIdbGroupName),
			scaleTopologyToString(ctx, eventBody.ExtraInfo.TopologyAfter, getTIdbGroupName),
		)

	case eventutil.EventTypeApplying, eventutil.EventTypeApplied:
		summary = renderApplyClusterSummary(eventBody.EventType, eventBody.ExtraInfo)

	case eventutil.EventTypeResuming:
		if eventBody.ExtraInfo.PausePlanType == globaldedicatedv1.Cluster_PausePlan_NORMAL.String() {
			autoResumeClusterCountTotal.WithLabelValues(
				strconv.FormatUint(eventBody.ClusterID, 10),
				cluster.DisplayName,
				strconv.FormatUint(cluster.ProjectId, 10),
				strconv.FormatUint(cluster.OrganizationId, 10),
			).Inc()
			summary = "Auto resumed as the pause time exceeds 7 days."
		}

	case eventutil.EventTypeResumed:
		if eventBody.ExtraInfo.PausePlanType == globaldedicatedv1.Cluster_PausePlan_NORMAL.String() {
			summary = "Cluster successfully auto resumed."
		}

	case eventutil.EventTypeUpdatingAuditLogConfig, eventutil.EventTypeUpdatedAuditLogConfig:
		if eventBody.ExtraInfo.AuditLogConfig != "" {
			summary += fmt.Sprintf(
				"\nAudit log config: %s.",
				eventBody.ExtraInfo.AuditLogConfig,
			)
		}
	case eventutil.EventTypeShutdownUpgrade:
		summary += fmt.Sprintf(
			"Start to shutdown upgrade cluster from version %s to %s.",
			*eventBody.ExtraInfo.OldVersion,
			*eventBody.ExtraInfo.NewVersion,
		)
	case eventutil.EventTypeShutdownUpgraded:
		summary += fmt.Sprintf(
			"Cluster successfully shutdown upgraded to version %s.",
			*eventBody.ExtraInfo.NewVersion,
		)
	}

	logger.Info("send to display event", zap.String("eventType", string(eventBody.EventType)))

	rewrittenEv := &tieventbus.Event{
		ID:   ev.ID + "rewrite",
		Type: event.DisplayEventTopic,
		Data: &event.DisplayEvent{
			DisplayEventType: displayMeta.DisplayEventType,
			DisplayStatus:    displayMeta.Status,
			Severity:         event.EventSeverityInfo,
			Summary:          summary,
			AuthorID:         eventBody.OperatorID,
			HappenedAtMs:     uint64(eventBody.EventCreatedAtMs),
			TenantID:         eventBody.TenantID,
			ProjectID:        eventBody.ProjectID,
			ClusterID:        eventBody.ClusterID,
			Detail:           eventBody,
		},
	}

	return c.eventbus.Publish(rewrittenEv)
}

func displayMetaForClusterEvent(eventBody *eventutil.ClusterEvent) (eventDisplayMeta, bool) {
	summary, exist := clusterEventSummaryMap[eventBody.EventType]
	if !exist {
		return eventDisplayMeta{}, false
	}

	meta := eventDisplayMeta{
		Summary:          summary,
		DisplayEventType: clusterEventTypeMap[eventBody.EventType],
		Status:           clusterEventStatusMap[eventBody.EventType],
	}

	if !isModifyClusterVolumeEvent(eventBody) {
		return meta, true
	}

	meta.DisplayEventType = string(eventutil.OperationSourceModifyClusterVolume)
	switch eventBody.EventType {
	case eventutil.EventTypeScaling:
		meta.Summary = "Start to modify cluster volume."
	case eventutil.EventTypeScaled:
		meta.Summary = "Cluster volume successfully modified."
	}
	if details := renderModifyClusterVolumeDetails(eventBody.ExtraInfo.OperationContext); details != "" {
		meta.Summary += "\nRequested changes: " + details + "."
	}
	return meta, true
}

func isModifyClusterVolumeEvent(eventBody *eventutil.ClusterEvent) bool {
	if eventBody == nil || eventBody.ExtraInfo == nil || eventBody.ExtraInfo.OperationContext == nil {
		return false
	}
	if eventBody.EventType != eventutil.EventTypeScaling && eventBody.EventType != eventutil.EventTypeScaled {
		return false
	}
	return eventBody.ExtraInfo.OperationContext.SourceOperation == string(eventutil.OperationSourceModifyClusterVolume)
}

func renderModifyClusterVolumeDetails(operationContext *eventutil.OperationContext) string {
	if operationContext == nil {
		return ""
	}

	requestedFields := requestedFieldSet(operationContext.RequestedFields)
	var changes []string
	for _, component := range operationContext.RequestedComponents {
		if component == nil || len(component.Storages) == 0 {
			continue
		}

		storageKeys := make([]string, 0, len(component.Storages))
		for diskType := range component.Storages {
			storageKeys = append(storageKeys, diskType)
		}
		slices.Sort(storageKeys)

		for _, diskType := range storageKeys {
			storage := component.Storages[diskType]
			if storage == nil {
				continue
			}

			var fields []string
			if requestedFields["storage_size"] {
				fields = append(fields, fmt.Sprintf("size %dGiB", storage.SizeGi))
			}
			if requestedFields["storage_iops"] {
				fields = append(fields, fmt.Sprintf("IOPS %d", storage.Iops))
			}
			if requestedFields["storage_throughput"] {
				fields = append(fields, fmt.Sprintf("throughput %dMiB/s", storage.ThroughputMibps))
			}
			if requestedFields["volume_type"] && storage.VolumeType != "" {
				fields = append(fields, fmt.Sprintf("volume type %s", storage.VolumeType))
			}
			if requestedFields["additional_iops"] {
				fields = append(fields, fmt.Sprintf("additional IOPS %d", storage.AdditionalIops))
			}
			if len(fields) == 0 {
				continue
			}

			displayDiskType := diskType
			if storage.DiskType != "" {
				displayDiskType = storage.DiskType
			}
			changes = append(changes, fmt.Sprintf(
				"%s %s: %s",
				displayComponentType(component.ComponentType),
				displayStorageDiskType(displayDiskType),
				strings.Join(fields, ", "),
			))
		}
	}

	return strings.Join(changes, "; ")
}

func renderApplyClusterSummary(eventType eventutil.EventType, extraInfo *eventutil.EventExtraInfo) string {
	var summary string
	switch eventType {
	case eventutil.EventTypeApplying:
		summary = "Start to apply cluster configuration."
		if extraInfo != nil && extraInfo.OldVersion != nil && extraInfo.NewVersion != nil {
			summary = fmt.Sprintf(
				"Start to upgrade cluster from version %s to %s.",
				*extraInfo.OldVersion,
				*extraInfo.NewVersion,
			)
		}
	case eventutil.EventTypeApplied:
		summary = "Cluster configuration successfully applied."
		if extraInfo != nil && extraInfo.OldVersion != nil && extraInfo.NewVersion != nil {
			summary = fmt.Sprintf(
				"Cluster successfully upgraded from version %s to %s.",
				*extraInfo.OldVersion,
				*extraInfo.NewVersion,
			)
		}
	default:
		return ""
	}

	if extraInfo == nil {
		return summary
	}
	if details := renderApplyOperationDetails(extraInfo.OperationContext); details != "" {
		summary += "\n" + details + "."
	}
	return summary
}

func renderApplyOperationDetails(operationContext *eventutil.OperationContext) string {
	if operationContext == nil ||
		operationContext.SourceOperation != string(eventutil.OperationSourceApplyCluster) {
		return ""
	}

	applyMask := operationContext.ApplyMask
	if applyMask == nil {
		return ""
	}

	var changes []string
	if applyMask.Version {
		changes = append(changes, "version")
	}
	for _, componentMask := range applyMask.ComponentApplyMasks {
		if componentMask == nil {
			continue
		}

		var componentChanges []string
		if componentMask.Configuration {
			componentChanges = append(componentChanges, "configuration")
		}
		if componentMask.Image {
			componentChanges = append(componentChanges, "image")
		}
		if len(componentChanges) == 0 {
			continue
		}

		changes = append(changes, fmt.Sprintf(
			"%s %s",
			displayComponentType(componentMask.ComponentType),
			strings.Join(componentChanges, " and "),
		))
	}
	if applyMask.Strategy != "" {
		changes = append(changes, fmt.Sprintf("strategy %s", applyMask.Strategy))
	}
	if applyMask.HotReload {
		changes = append(changes, "hot reload")
	}

	if len(changes) == 0 && len(operationContext.RequestedFields) > 0 {
		changes = append(changes, strings.Join(operationContext.RequestedFields, ", "))
	}
	if len(changes) == 0 {
		return ""
	}
	return "Requested changes: " + strings.Join(changes, "; ")
}

func requestedFieldSet(fields []string) map[string]bool {
	out := make(map[string]bool, len(fields))
	for _, field := range fields {
		out[field] = true
	}
	return out
}

func displayComponentType(componentType string) string {
	switch componentType {
	case globaldedicatedv1.ComponentType_TIDB.String():
		return "TiDB"
	case globaldedicatedv1.ComponentType_TIPROXY.String():
		return "TiProxy"
	case globaldedicatedv1.ComponentType_TIKV.String():
		return "TiKV"
	case globaldedicatedv1.ComponentType_TIFLASH.String():
		return "TiFlash"
	case globaldedicatedv1.ComponentType_PD.String():
		return "PD"
	default:
		return componentType
	}
}

func displayStorageDiskType(diskType string) string {
	switch diskType {
	case globaldedicatedv1.Sku_DATA.String():
		return "data disk"
	case globaldedicatedv1.Sku_RAFT_LOG.String():
		return "raft log disk"
	default:
		return strings.ToLower(strings.ReplaceAll(diskType, "_", " "))
	}
}

func (c *clusterEventConsumerManagerImp) loadTopologyFromReservedCapacityTemplate(
	ctx context.Context,
	cluster *entity.Cluster,
	eventBody *eventutil.ClusterEvent,
) *eventutil.ClusterTopology {
	nextGenCluster, err := cluster.GetNextGenCluster()
	if err != nil {
		tilog.Warn(ctx, "failed to parse nextgen cluster for reserved capacity template topology", zap.Error(err))
		return nil
	}

	templateID := nextGenCluster.GetReservedCapacityTemplateId()
	if templateID == 0 {
		return nil
	}

	template, err := c.nextgenCli.GetReservedCapacityTemplate(ctx, &globalngv1.GetReservedCapacityTemplateRequest{CapacityTemplateId: templateID})
	if err != nil {
		tilog.Warn(ctx,
			"failed to load reserved capacity template for topology",
			zap.Uint64("cluster_id", eventBody.ClusterID),
			zap.Uint64("template_id", templateID),
			zap.Error(err),
		)
		return nil
	}

	topology := convertReservedTemplateTopology(template.GetFixedTopology())
	return topology
}

func convertReservedTemplateTopology(fixedTopology *globalngv1.FixedTopology) *eventutil.ClusterTopology {
	if fixedTopology == nil {
		return nil
	}

	components := make(map[string]*eventutil.ClusterComponent)
	for _, component := range fixedTopology.GetComponents() {
		if component == nil {
			continue
		}

		componentType := reservedTemplateComponentTypeToDisplayComponentType(component.GetComponentType())
		if componentType == "" {
			continue
		}

		nodeQuota := component.GetNodeQuota()
		if nodeQuota == nil || nodeQuota.GetNodeCount() == 0 {
			continue
		}

		cc := &eventutil.ClusterComponent{
			ComponentType: componentType,
			Quantity:      nodeQuota.GetNodeCount(),
		}

		nodeSpec := nodeQuota.GetNodeSpec()
		if nodeSpec != nil {
			cc.SpuName = fmt.Sprintf("%dC%dG", nodeSpec.GetCpu(), nodeSpec.GetMemoryGi())
			cc.SpuDisplayName = cc.SpuName
		}

		for _, storage := range component.GetStorages() {
			if storage == nil {
				continue
			}
			switch storage.GetName() {
			case globalngv1.StorageSpec_DATA:
				cc.StorageSizeGi = storage.GetSizeGi()
				cc.VolumeType = storage.GetType()
				cc.AdditionalIops = storage.GetIops()
			case globalngv1.StorageSpec_RAFTLOG:
				cc.RaftLogVolumeType = storage.GetType()
				cc.RaftLogIops = storage.GetIops()
			}
		}

		if existing, ok := components[componentType]; ok {
			mergeClusterComponent(existing, cc)
		} else {
			components[componentType] = cc
		}
	}

	if len(components) == 0 {
		return nil
	}

	return &eventutil.ClusterTopology{Components: components}
}

func mergeClusterComponent(dst, src *eventutil.ClusterComponent) {
	if dst == nil || src == nil {
		return
	}

	dst.Quantity += src.Quantity

	if dst.SpuName == "" {
		dst.SpuName = src.SpuName
		dst.SpuDisplayName = src.SpuDisplayName
	}

	if dst.StorageSizeGi == 0 {
		dst.StorageSizeGi = src.StorageSizeGi
	}
	if dst.VolumeType == "" {
		dst.VolumeType = src.VolumeType
	}
	if dst.AdditionalIops == 0 {
		dst.AdditionalIops = src.AdditionalIops
	}
	if dst.RaftLogVolumeType == "" {
		dst.RaftLogVolumeType = src.RaftLogVolumeType
	}
	if dst.RaftLogIops == 0 {
		dst.RaftLogIops = src.RaftLogIops
	}
}

func reservedTemplateComponentTypeToDisplayComponentType(componentType globalngv1.ComponentType) string {
	switch componentType {
	case globalngv1.ComponentType_TIDB:
		return globaldedicatedv1.ComponentType_TIDB.String()
	case globalngv1.ComponentType_TIPROXY:
		return globaldedicatedv1.ComponentType_TIPROXY.String()
	case globalngv1.ComponentType_TIKV:
		return globaldedicatedv1.ComponentType_TIKV.String()
	case globalngv1.ComponentType_TIFLASH_WRITE, globalngv1.ComponentType_TIFLASH_COMPUTE:
		return globaldedicatedv1.ComponentType_TIFLASH.String()
	default:
		return ""
	}
}

// sendResourcePoolToDisplayEvent converts resource pool cluster events into display events.
func (c *clusterEventConsumerManagerImp) sendResourcePoolToDisplayEvent(ctx context.Context, ev *tieventbus.Event) error {
	logger := tilog.WithContextS(ctx)

	eventBody, ok := ev.Data.(*eventutil.ClusterEvent)
	if !ok {
		logger.Errorf("event id %s is not a ClusterEvent", ev.ID)
		return nil
	}

	summary, exist := resourcePoolEventSummaryMap[eventBody.EventType]
	if !exist {
		tilog.Info(ctx, "skip unhandled resource pool event type",
			zap.String("eventType", string(eventBody.EventType)))
		return nil
	}

	logger = logger.With(zap.Uint64("resourcePoolID", eventBody.ClusterID))
	logger.Infof(
		"cluster display event consumer, try to handle resource pool event, id=%s, poolID=%d, eventType=%s",
		ev.ID, eventBody.ClusterID, eventBody.EventType,
	)

	// Enrich summary with resource pool details.
	switch eventBody.EventType {
	case eventutil.EventTypeCreated:
		summary = fmt.Sprintf("Resource pool successfully created. Resource pool ID: %d.", eventBody.ClusterID)
		if eventBody.ExtraInfo != nil && eventBody.ExtraInfo.ResourcePoolTopology != "" {
			summary += fmt.Sprintf(" Topology: %s", eventBody.ExtraInfo.ResourcePoolTopology)
		}

	case eventutil.EventTypeDeleting:
		summary = fmt.Sprintf("Start to delete resource pool (resource pool id: %d).", eventBody.ClusterID)

	case eventutil.EventTypeDeleted:
		summary = fmt.Sprintf("Resource pool %d successfully deleted.", eventBody.ClusterID)

	case eventutil.EventTypeResourcePoolDefaultSet:
		summary = fmt.Sprintf("Successfully set the default resource pool. Tenant ID: %d, Resource Pool ID: %d",
			eventBody.TenantID, eventBody.ClusterID)
		if eventBody.ExtraInfo != nil {
			if eventBody.ExtraInfo.Cloud != "" {
				summary += fmt.Sprintf(", Cloud: %s", eventBody.ExtraInfo.Cloud)
			}
			if eventBody.ExtraInfo.Region != "" {
				summary += fmt.Sprintf(", Region: %s.", eventBody.ExtraInfo.Region)
			} else {
				summary += "."
			}
		} else {
			summary += "."
		}

	case eventutil.EventTypeResourcePoolInstanceLimitUpdated:
		if eventBody.ExtraInfo != nil &&
			eventBody.ExtraInfo.InstanceLimitBefore != nil &&
			eventBody.ExtraInfo.InstanceLimitAfter != nil {
			summary = fmt.Sprintf(
				"Successfully updated instance limit. Resource Pool ID: %d. Before: {instance_limit: %d}, After: {instance_limit: %d}.",
				eventBody.ClusterID,
				*eventBody.ExtraInfo.InstanceLimitBefore,
				*eventBody.ExtraInfo.InstanceLimitAfter,
			)
		} else {
			summary = fmt.Sprintf("Successfully updated instance limit. Resource Pool ID: %d.", eventBody.ClusterID)
		}

	case eventutil.EventTypeInstanceCreatingOnResourcePool:
		if eventBody.ExtraInfo != nil && eventBody.ExtraInfo.InstanceID != nil {
			summary = fmt.Sprintf("Start to create instance %d on resource pool %d.", *eventBody.ExtraInfo.InstanceID, eventBody.ClusterID)
		} else {
			summary = fmt.Sprintf("Start to create instance on resource pool %d.", eventBody.ClusterID)
		}

	case eventutil.EventTypeInstanceCreatedOnResourcePool:
		if eventBody.ExtraInfo != nil && eventBody.ExtraInfo.InstanceID != nil {
			summary = fmt.Sprintf("Instance %d successfully created on resource pool %d.", *eventBody.ExtraInfo.InstanceID, eventBody.ClusterID)
		} else {
			summary = fmt.Sprintf("Instance successfully created on resource pool %d.", eventBody.ClusterID)
		}

	case eventutil.EventTypeInstanceDeletingOnResourcePool:
		if eventBody.ExtraInfo != nil && eventBody.ExtraInfo.InstanceID != nil {
			summary = fmt.Sprintf("Start to delete instance %d from resource pool %d.", *eventBody.ExtraInfo.InstanceID, eventBody.ClusterID)
		} else {
			summary = fmt.Sprintf("Start to delete instance from resource pool %d.", eventBody.ClusterID)
		}

	case eventutil.EventTypeInstanceDeletedOnResourcePool:
		if eventBody.ExtraInfo != nil && eventBody.ExtraInfo.InstanceID != nil {
			summary = fmt.Sprintf("Instance %d successfully deleted from resource pool %d.", *eventBody.ExtraInfo.InstanceID, eventBody.ClusterID)
		} else {
			summary = fmt.Sprintf("Instance successfully deleted from resource pool %d.", eventBody.ClusterID)
		}
	}

	logger.Info("send resource pool to display event", zap.String("eventType", string(eventBody.EventType)))

	rewrittenEv := &tieventbus.Event{
		ID:   ev.ID + "rewrite",
		Type: event.DisplayEventTopic,
		Data: &event.DisplayEvent{
			DisplayEventType: resourcePoolEventTypeMap[eventBody.EventType],
			DisplayStatus:    resourcePoolEventStatusMap[eventBody.EventType],
			Severity:         event.EventSeverityInfo,
			Summary:          summary,
			AuthorID:         eventBody.OperatorID,
			HappenedAtMs:     uint64(eventBody.EventCreatedAtMs),
			TenantID:         eventBody.TenantID,
			ProjectID:        eventBody.ProjectID,
			ClusterID:        eventBody.ClusterID,
			Detail:           eventBody,
		},
	}

	return c.eventbus.Publish(rewrittenEv)
}

// sendInstanceToDisplayEvent converts nextgen instance cluster events into display events.
func (c *clusterEventConsumerManagerImp) sendInstanceToDisplayEvent(ctx context.Context, ev *tieventbus.Event) error {
	logger := tilog.WithContextS(ctx)

	eventBody, ok := ev.Data.(*eventutil.ClusterEvent)
	if !ok {
		logger.Errorf("event id %s is not a ClusterEvent", ev.ID)
		return nil
	}

	summary, exist := instanceEventSummaryMap[eventBody.EventType]
	if !exist {
		// Skip unhandled event types (e.g. EventTypeInitializing).
		return nil
	}

	logger = logger.With(zap.Uint64("clusterID", eventBody.ClusterID))
	logger.Infof(
		"cluster display event consumer, try to handle instance event, id=%s, clusterID=%d, eventType=%s",
		ev.ID, eventBody.ClusterID, eventBody.EventType,
	)

	// Attempt to fetch the cluster to enrich the summary with instance-specific data.
	cluster, err := c.clusterRepo.GetCluster(ctx, eventBody.ClusterID, true)
	if err != nil {
		if errors.Is(err, gorm.ErrRecordNotFound) {
			logger.Infof("ignore instance event: cluster %d not found in mgmt service", eventBody.ClusterID)
			return nil
		}
		logger.Error("failed to get cluster", zap.Error(err))
		return err
	}

	nextGenCluster, ngErr := cluster.GetNextGenCluster()

	switch eventBody.EventType {
	case eventutil.EventTypeCreating:
		if ngErr == nil && nextGenCluster.GetSharedPoolId() != "" {
			summary = fmt.Sprintf("Start to create instance on resource pool %s.", nextGenCluster.GetSharedPoolId())
		}
	case eventutil.EventTypeCreated:
		if ngErr == nil {
			summary = fmt.Sprintf(
				"The instance was successfully created on resource pool %s. instance id: %d, MaxRCU: %d.",
				nextGenCluster.GetSharedPoolId(),
				eventBody.ClusterID,
				nextGenCluster.GetMaxRcu(),
			)
		}
	case eventutil.EventTypeUpdatedCapacity:
		if ngErr == nil {
			if eventBody.ExtraInfo != nil && eventBody.ExtraInfo.OldMaxRcu != nil && eventBody.ExtraInfo.NewMaxRcu != nil {
				summary = fmt.Sprintf(
					"Instance capacity successfully updated. instance id: %d, MaxRCU: %d -> %d.",
					eventBody.ClusterID,
					*eventBody.ExtraInfo.OldMaxRcu,
					*eventBody.ExtraInfo.NewMaxRcu,
				)
			} else {
				summary = fmt.Sprintf(
					"Instance capacity successfully updated. instance id: %d, MaxRCU: %d.",
					eventBody.ClusterID,
					nextGenCluster.GetMaxRcu(),
				)
			}
		}
	case eventutil.EventTypeDeleting:
		summary = fmt.Sprintf("Start to delete instance %d.", eventBody.ClusterID)
	case eventutil.EventTypeDeleted:
		summary = fmt.Sprintf("Instance %d successfully deleted.", eventBody.ClusterID)
	}

	logger.Info("send instance to display event", zap.String("eventType", string(eventBody.EventType)))

	rewrittenEv := &tieventbus.Event{
		ID:   ev.ID + "rewrite",
		Type: event.DisplayEventTopic,
		Data: &event.DisplayEvent{
			DisplayEventType: instanceEventTypeMap[eventBody.EventType],
			DisplayStatus:    instanceEventStatusMap[eventBody.EventType],
			Severity:         event.EventSeverityInfo,
			Summary:          summary,
			AuthorID:         eventBody.OperatorID,
			HappenedAtMs:     uint64(eventBody.EventCreatedAtMs),
			TenantID:         eventBody.TenantID,
			ProjectID:        eventBody.ProjectID,
			ClusterID:        eventBody.ClusterID,
			Detail:           eventBody,
		},
	}

	return c.eventbus.Publish(rewrittenEv)
}

func isNextGenInstancePlan(servicePlan string) bool {
	switch servicePlan {
	case "Premium", "BYOC", "Premium_Reserved", "Essential_V2":
		return true
	default:
		return false
	}
}

func createTopologyToString(globalCluster *globaldedicatedv1.Cluster) string {
	var result []string

	// do not show pd to user

	tidb, ok := lo.Find(
		globalCluster.GetClusterComponents(),
		func(comp *globaldedicatedv1.Cluster_ClusterComponent) bool {
			return comp.GetComponentType() == globaldedicatedv1.ComponentType_TIDB
		},
	)
	if ok {
		eventStr := renderTidbTopologyStr(tidb.GetCount(), tidb.GetSpuName())
		tiproxy, ok := lo.Find(
			globalCluster.GetClusterComponents(),
			func(comp *globaldedicatedv1.Cluster_ClusterComponent) bool {
				return comp.GetComponentType() == globaldedicatedv1.ComponentType_TIPROXY
			},
		)
		if ok {
			eventStr += " and "
			eventStr += renderTiproxyTopologyStr(tiproxy.GetCount(), tiproxy.GetSpuDisplayName())
		}
		eventStr += " in DefaultGroup"
		result = append(result, eventStr)
	}

	tikv, ok := lo.Find(
		globalCluster.GetClusterComponents(),
		func(comp *globaldedicatedv1.Cluster_ClusterComponent) bool {
			return comp.GetComponentType() == globaldedicatedv1.ComponentType_TIKV
		},
	)
	if ok {
		dataVt := tikv.GetStorages()[globaldedicatedv1.Sku_DATA.String()].GetVolumeType()
		raftVt := tikv.GetStorages()[globaldedicatedv1.Sku_RAFT_LOG.String()].GetVolumeType()
		st := storagetype.BuildCurrentTikvStorageType(dataVt, raftVt)
		var dataIops *uint32
		if st == openapidedicatedv1beta1.Cluster_StorageNodeSetting_Plus {
			dataIops = lo.EmptyableToPtr(tikv.GetStorages()[globaldedicatedv1.Sku_DATA.String()].GetIops())
		}
		var raftIops *uint32
		if st == openapidedicatedv1beta1.Cluster_StorageNodeSetting_Performance {
			raftIops = lo.ToPtr(tikv.GetStorages()[globaldedicatedv1.Sku_RAFT_LOG.String()].GetIops())
		}
		result = append(result, renderTikvTopologyStr(
			tikv.GetCount(),
			tikv.GetStorages()[globaldedicatedv1.Sku_DATA.String()].GetSizeGi(),
			tikv.GetSpuName(),
			st.String(),
			dataIops,
			raftIops,
			lo.EmptyableToPtr(tikv.GetStorages()[globaldedicatedv1.Sku_DATA.String()].GetAdditionalIops()),
		))
	}

	tiflash, ok := lo.Find(
		globalCluster.GetClusterComponents(),
		func(comp *globaldedicatedv1.Cluster_ClusterComponent) bool {
			return comp.GetComponentType() == globaldedicatedv1.ComponentType_TIFLASH
		},
	)
	if ok && tiflash.GetCount() > 0 {
		dataVt := tiflash.GetStorages()[globaldedicatedv1.Sku_DATA.String()].GetVolumeType()
		result = append(result, renderTiflashTopologyStr(
			tiflash.GetCount(),
			tiflash.GetStorages()[globaldedicatedv1.Sku_DATA.String()].GetSizeGi(),
			tiflash.GetSpuName(),
			storagetype.BuildCurrentTiflashStorageType(dataVt, globaldedicatedv1.StorageMu_VOLUME_TYPE_UNSPECIFIED).String(),
		),
		)
	}

	return strings.Join(result, " | ")
}

func scaleTopologyToString(
	ctx context.Context,
	topology *eventutil.ClusterTopology,
	getTidbNodeGroupName func(tidbNodeGroup *eventutil.TidbNodeGroup) (string, error),
) string {
	var result []string

	components := topology.Components

	// do not show pd to user

	if _, ok := components[globaldedicatedv1.ComponentType_TIDB.String()]; ok {
		result = append(result, tidbScaleTopologyToString(ctx, topology, getTidbNodeGroupName))
	}

	if tikv, ok := components[globaldedicatedv1.ComponentType_TIKV.String()]; ok {
		dataVt, err := enum.FromString[globaldedicatedv1.StorageMu_VolumeType](tikv.VolumeType)
		if err != nil {
			tilog.Error(ctx, "load tikv volume type failed", zap.String("volumeType", tikv.VolumeType), zap.Error(err))
		}
		raftVt, err := enum.FromString[globaldedicatedv1.StorageMu_VolumeType](tikv.RaftLogVolumeType)
		if err != nil {
			tilog.Error(
				ctx,
				"load tikv raft log volume type failed",
				zap.String("raftLogVolumeType", tikv.RaftLogVolumeType),
				zap.Error(err),
			)
		}
		st := storagetype.BuildCurrentTikvStorageType(
			globaldedicatedv1.StorageMu_VolumeType(dataVt),
			globaldedicatedv1.StorageMu_VolumeType(raftVt),
		)
		var dataIops *uint32
		if st == openapidedicatedv1beta1.Cluster_StorageNodeSetting_Plus {
			dataIops = lo.EmptyableToPtr(tikv.StorageIops)
		}
		var raftIops *uint32
		if st == openapidedicatedv1beta1.Cluster_StorageNodeSetting_Performance {
			raftIops = lo.ToPtr(tikv.RaftLogIops)
		}
		result = append(result, renderTikvTopologyStr(
			tikv.Quantity,
			tikv.StorageSizeGi,
			tikv.SpuName,
			st.String(),
			dataIops,
			raftIops,
			lo.EmptyableToPtr(tikv.AdditionalIops),
		))
	}

	// tiflash may not exist
	if tiflash, ok := components[globaldedicatedv1.ComponentType_TIFLASH.String()]; ok && tiflash.Quantity > 0 {
		dataVt, err := enum.FromString[globaldedicatedv1.StorageMu_VolumeType](tiflash.VolumeType)
		if err != nil {
			tilog.Error(
				ctx,
				"load tiflash volume type failed",
				zap.String("volumeType", tiflash.VolumeType),
				zap.Error(err),
			)
		}
		result = append(result, renderTiflashTopologyStr(
			tiflash.Quantity,
			tiflash.StorageSizeGi,
			tiflash.SpuName,
			storagetype.BuildCurrentTiflashStorageType(globaldedicatedv1.StorageMu_VolumeType(dataVt), globaldedicatedv1.StorageMu_VOLUME_TYPE_UNSPECIFIED).
				String(),
		))
	}

	return strings.Join(result, " | ")
}

func tidbScaleTopologyToString(
	ctx context.Context,
	topology *eventutil.ClusterTopology,
	getTidbNodeGroupName func(tidbNodeGroup *eventutil.TidbNodeGroup) (string, error),
) string {
	tidb := topology.Components[globaldedicatedv1.ComponentType_TIDB.String()]
	tiproxy := topology.Components[globaldedicatedv1.ComponentType_TIPROXY.String()]
	eventStr := renderTidbTopologyStr(tidb.Quantity, tidb.SpuName)

	if tiproxy != nil && tiproxy.Quantity > 0 {
		eventStr += " and "
		eventStr += renderTiproxyTopologyStr(tiproxy.Quantity, tiproxy.SpuDisplayName)
	}

	eventStr += " in DefaultGroup"

	if len(topology.ExTidbNodeGroups) == 0 {
		return eventStr
	}

	result := []string{eventStr}
	for _, tidbNodeGroup := range topology.ExTidbNodeGroups {
		tidbGroupName, err := getTidbNodeGroupName(tidbNodeGroup)
		if err != nil {
			tilog.Error(ctx, "load tidb node group failed", zap.Uint64("id", tidbNodeGroup.ID), zap.Error(err))
			tidbGroupName = fmt.Sprintf("%d", tidbNodeGroup.ID)
		}

		s := renderTidbTopologyStr(tidbNodeGroup.TidbNodeQuantity, tidb.SpuName)
		if tidbNodeGroup.TiProxyQuantity > 0 {
			s += " and "
			s += renderTiproxyTopologyStr(tidbNodeGroup.TiProxyQuantity, tidbNodeGroup.TiProxySpuDisplayName)
		}

		result = append(result, fmt.Sprintf("%s in %q(%d)", s, tidbGroupName, tidbNodeGroup.ID))
	}

	return strings.Join(result, " | ")
}

func (c *clusterEventConsumerManagerImp) sendClusterMaintenanceDisplayEvent(
	ctx context.Context,
	ev *tieventbus.Event,
) error {
	logger := tilog.WithContextS(ctx)
	eventBody, ok := ev.Data.(*eventutil.MaintenanceWindowEvent)
	if !ok {
		logger.Errorf("event id %s is not a MaintenanceWindowEvent", ev.ID)
		return nil
	}

	switch eventBody.EventType {
	case eventutil.MaintenanceWindowEventTypeExecutionRunning,
		eventutil.MaintenanceWindowEventTypeExecutionDone,
		eventutil.MaintenanceWindowEventTypeExecutionCanceled:
		// continue
	default:
		// ignore other events
		return nil
	}

	execution := eventBody.MaintenanceTaskExecution
	if execution == nil {
		logger.Errorf("event id %s has no task execution field", ev.ID)
		// log error and return nil, as we can not handle the event and it's not retryable
		return nil
	}

	logger = logger.With(zap.Uint64("clusterID", execution.ClusterID))
	logger.Infof(
		"clinic consumer, try to handle event, id=%s, clusterID=%d, eventType=%s",
		ev.ID,
		execution.ClusterID,
		eventBody.EventType,
	)

	task, err := c.maintenanceTaskRepo.Load(ctx, execution.MaintenanceTaskID)
	if err != nil {
		logger.Info("failed to get maintenance task, id=%d, err=%s", execution.MaintenanceTaskID, err)
		return fmt.Errorf("failed to get maintenance task, id=%d, err=%w", execution.MaintenanceTaskID, err)
	}

	project, rawErr := c.accountCli.GetProject(ctx, eventBody.ProjectID)
	if rawErr != nil {
		logger.Info("failed to get project, id=%d, err=%s", eventBody.ProjectID, rawErr)
		return fmt.Errorf("failed to get project, id=%d, err=%w", eventBody.ProjectID, rawErr)
	}

	summary := fmt.Sprintf("%s Task: %s", maintenanceWindowEventSummaryMap[eventBody.EventType], task.Description)

	rewrittenEv := &tieventbus.Event{
		ID:   ev.ID + "rewrite",
		Type: event.DisplayEventTopic,
		Data: &event.DisplayEvent{
			DisplayEventType: maintenanceWindowEventTypeMap[eventBody.EventType],
			DisplayStatus:    maintenanceWindowEventStatusMap[eventBody.EventType],
			Severity:         event.EventSeverityInfo,
			Summary:          summary,
			AuthorID:         0,
			HappenedAtMs:     uint64(eventBody.EventCreatedAtMs),
			TenantID:         project.TenantId,
			ProjectID:        eventBody.ProjectID,
			ClusterID:        execution.ClusterID,
			Detail:           eventBody,
		},
	}
	return c.eventbus.Publish(rewrittenEv)
}