package consumer

import (
	"context"
	"fmt"
	"sync"

	"github.com/pingcap/errors"
	"github.com/tidbcloud/metrics-integration/global/internal/infrastructure/grpc/grpcli"
	"github.com/tidbcloud/metrics-integration/global/pkg/log"
	clusterevent "github.com/tidbcloud/tidb-management-service/utilities/clusterevent"
	"github.com/tidbcloud/tiinf/transport/tieventbus"
	"go.uber.org/multierr"
	"go.uber.org/zap"

	mgmt "github.com/tidbcloud/tidb-management-service/api/spec/tidb_mgmt_service/v1"
)

const (
	CMEventType = "ClusterEvent"
)

type CMEventHandler interface {
	Name() string
	Match(event *clusterevent.ClusterEvent) bool
	Execute(ctx context.Context, event *clusterevent.ClusterEvent, cluster *mgmt.Cluster) error
}

type CMEventHandlerDispatcher interface {
	Register(handler CMEventHandler)
	Execute(event *clusterevent.ClusterEvent) error
}

func GetCMEventHandlerDispatcher() CMEventHandlerDispatcher {
	return cmEventHandlerSerialDispatcherSingleton
}

type CMEventHandlerSerialDispatcher struct {
	cmEventHandler map[string]CMEventHandler
	lck            sync.Mutex
}

var cmEventHandlerSerialDispatcherSingleton = &CMEventHandlerSerialDispatcher{}

var getClusterByID = grpcli.GetClusterByID

func (c *CMEventHandlerSerialDispatcher) Register(handler CMEventHandler) {
	c.lck.Lock()
	defer c.lck.Unlock()

	if c.cmEventHandler == nil {
		c.cmEventHandler = make(map[string]CMEventHandler, 64)
	}
	c.cmEventHandler[handler.Name()] = handler
}

func (c *CMEventHandlerSerialDispatcher) Execute(event *clusterevent.ClusterEvent) error {
	ctx := log.WithLogID(context.Background())
	logger := log.GetLogger(ctx)
	logger.Info("cmEvent", zap.Any("message", event))

	var matchedHandlers []CMEventHandler
	for _, handler := range c.cmEventHandler {
		if handler.Match(event) {
			matchedHandlers = append(matchedHandlers, handler)
		}
	}
	if len(matchedHandlers) == 0 {
		logger.Info("skip CM event, no handler matched", zap.Any("message", event))
		return nil
	}

	// get infos from cluster service
	cluster, err := getClusterByID(ctx, uint64(event.ClusterID))
	if err != nil {
		// For NextGen/TiDB-X clusters (Starter/Essential/Premium/BYOC), the mgmt API
		// may not enumerate them the same way as Dedicated clusters. If the event carries
		// ServicePlan or ClusterType (fields only present in tidb-mgmt events), we fall
		// back to a minimal cluster object so that handlers can still act on the event.
		if event.ServicePlan != "" || event.ClusterType != "" {
			logger.Warn("integration event: GetClusterByID failed for non-dedicated cluster, proceeding with event-derived info",
				zap.Uint64("clusterID", event.ClusterID),
				zap.String("servicePlan", event.ServicePlan),
				zap.String("clusterType", string(event.ClusterType)),
				zap.Error(err),
			)
			cluster = &mgmt.Cluster{ClusterId: event.ClusterID}
		} else {
			logger.Error("integration event: GetAnyClusterByID error",
				zap.Uint64("clusterID", event.ClusterID),
				zap.Error(err),
			)
			return fmt.Errorf("failed to get cluster info: %w", err)
		}
	}

	var errs error
	for _, handler := range matchedHandlers {
		logger.Info("consume CM event",
			zap.String("handler name", handler.Name()),
			zap.Any("message", event))
		if err := handler.Execute(ctx, event, cluster); err != nil {
			warpErr := errors.Annotate(err, fmt.Sprintf("cluster event: %s, handler name: %s",
				event.EventType, handler.Name()))
			errs = multierr.Append(errs, warpErr)
		}
	}
	if errs != nil {
		logger.Error("consumer CM event occur error!", zap.Error(errs))
	}
	return errs
}

func buildCMEventConsumer() tieventbus.HandlerConfig {
	return tieventbus.HandlerConfig{
		EventType: CMEventType,
		Group:     group,
		HandleFunc: func(event *tieventbus.Event) error {
			cmEvent := event.Data.(*clusterevent.ClusterEvent)
			return GetCMEventHandlerDispatcher().Execute(cmEvent)
		},
		DataStruct: clusterevent.ClusterEvent{},
	}
}