package consumer

import (
	"context"
	"fmt"
	"sync"

	"github.com/pingcap/errors"
	core_event "github.com/tidbcloud/core-svc/event"
	"github.com/tidbcloud/core-svc/idl/pbgen/proto/core"
	"github.com/tidbcloud/dataflow-service/app/container"
	"github.com/tidbcloud/dataflow-service/pkg/log"
	"github.com/tidbcloud/pkgv2/bizerr"
	tidb_mgmt_servicev1 "github.com/tidbcloud/tidb-management-service/api/spec/tidb_mgmt_service/v1"
	"github.com/tidbcloud/tiinf/transport/tieventbus"
	"go.uber.org/multierr"
	"go.uber.org/zap"
)

const (
	CMEventType = "ClusterEvent"
)

type CMEventHandler interface {
	Name() string
	// WatchForEvents returns the event types this handler is interested in.
	// The dispatcher uses this at registration time to build a watched-type set,
	// so that GetAnyClusterByID is only called when at least one handler may
	// process the event. WatchForEvents must be a superset of the types that
	// Match() can accept: if an event type is not listed here the event will be
	// silently dropped before Match() is ever evaluated.
	WatchForEvents() []core_event.ClusterEventType
	Match(event *core_event.ClusterEvent) bool
	Execute(ctx context.Context, event *core_event.ClusterEvent, cluster *core.Cluster) error
}

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

func GetCMEventHandlerDispatcher() CMEventHandlerDispatcher {
	return cmEventHandlerSerialDispatcherSingleton
}

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

var cmEventHandlerSerialDispatcherSingleton = &CMEventHandlerSerialDispatcher{}

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

	if c.cmEventHandler == nil {
		c.cmEventHandler = make(map[string]CMEventHandler, 64)
	}
	if c.watchedEventTypes == nil {
		c.watchedEventTypes = make(map[core_event.ClusterEventType]struct{}, 16)
	}
	c.cmEventHandler[handler.Name()] = handler
	for _, et := range handler.WatchForEvents() {
		c.watchedEventTypes[et] = struct{}{}
	}
}

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

	// Pre-filter: skip events that no registered handler has declared interest in.
	// This avoids calling GetAnyClusterByID (and potentially triggering DLQ alerts)
	// for events from cluster types (e.g. NextGen) that dataflow-service does not
	// manage.
	if _, watched := c.watchedEventTypes[event.EventType]; !watched {
		logger.Info("cmEvent: no handler watches this event type, skip",
			zap.String("eventType", string(event.EventType)),
			zap.Uint64("clusterID", event.ClusterID))
		return nil
	}

	app := container.Global()

	// Verify the cluster is a dedicated cluster before invoking handlers.
	// Non-dedicated clusters (Premium, BYOC, NG) may emit the same event types
	// that dedicated handlers watch (e.g. ClusterCreating, ClusterNormal), but
	// core-svc cannot look them up and GetAnyClusterByID would fail with code=49900004,
	// causing the event to be sent to DLQ.
	mgmtCluster, mgmtErr := app.MgmtSvcGrpcClient.GetAnyClusterByID(ctx, event.ClusterID)
	if mgmtErr != nil {
		if mgmtErr.ErrCode() == bizerr.ErrCodeRecordNotFound {
			logger.Warn("cmEvent: cluster not found in mgmt-svc, skip",
				zap.Uint64("clusterID", event.ClusterID),
				zap.String("eventType", string(event.EventType)))
			return nil
		}
		logger.Error("cmEvent: GetClusterByID error",
			zap.Uint64("clusterID", event.ClusterID),
			zap.Error(mgmtErr))
		return mgmtErr
	}
	if mgmtCluster.DeployType != tidb_mgmt_servicev1.DeployType_DEPLOY_TYPE_DEDICATED {
		logger.Info("cmEvent: not a dedicated cluster, skip",
			zap.Uint64("clusterID", event.ClusterID),
			zap.String("eventType", string(event.EventType)))
		return nil
	}

	cluster, err := app.CoreGrpcClient.GetAnyClusterByID(ctx, event.ClusterID)
	if err != nil {
		logger.Error("backup event: GetAnyClusterByID error",
			zap.Uint64("clusterID", event.ClusterID),
			zap.Error(err),
		)
		return err
	}

	var errs error
	for _, handler := range c.cmEventHandler {
		match := handler.Match(event)
		if !match {
			continue
		}
		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.(*core_event.ClusterEvent)
			return GetCMEventHandlerDispatcher().Execute(cmEvent)
		},
		DataStruct: core_event.ClusterEvent{},
	}
}