package clustereventconsumer

import (
	"context"
	"fmt"
	"net/url"

	"github.com/Shopify/sarama"
	"github.com/prometheus/client_golang/prometheus"
	"github.com/tidbcloud/pkgv2/db"
	"github.com/tidbcloud/tiinf/monitor/tilog"
	"github.com/tidbcloud/tiinf/transport/tieventbus"
	"go.uber.org/zap"
	"golang.org/x/net/proxy"

	"github.com/tidbcloud/tidb-management-service/tidb-mgmt-service/domain/adapter/account"
	"github.com/tidbcloud/tidb-management-service/tidb-mgmt-service/domain/adapter/billing"
	"github.com/tidbcloud/tidb-management-service/tidb-mgmt-service/domain/adapter/dedicatedglobal"
	"github.com/tidbcloud/tidb-management-service/tidb-mgmt-service/domain/adapter/jsm"
	"github.com/tidbcloud/tidb-management-service/tidb-mgmt-service/domain/adapter/nextgenglobal"
	"github.com/tidbcloud/tidb-management-service/tidb-mgmt-service/domain/adapter/tempcore"
	"github.com/tidbcloud/tidb-management-service/tidb-mgmt-service/domain/repository"
	infrarepo "github.com/tidbcloud/tidb-management-service/tidb-mgmt-service/infra/repository"
	"github.com/tidbcloud/tidb-management-service/tidb-mgmt-service/pkg/config"
	eventutil "github.com/tidbcloud/tidb-management-service/utilities/clusterevent"
)

const AppName = "tidb_management_service"

type Manager interface {
	Run() error
}

type clusterEventConsumerManagerImp struct {
	jsmClient           jsm.Client
	coreSvc             tempcore.Client
	clusterRepo         repository.ClusterRepository
	maintenanceTaskRepo repository.MaintenanceTaskRepo
	dedicatedCli        dedicatedglobal.DedicatedGlobal
	nextgenCli          nextgenglobal.NextGenGlobal

	billingCli       billing.Client
	accountCli       account.Client
	uncConfig        *config.UserNotificationCenter
	dbProvider       db.TransProvider
	customStatusRepo repository.CustomStatusRepository

	eventbus tieventbus.Eventbus
}

func NewManager(
	jsmClient jsm.Client,
	coreClient tempcore.Client,
	clusterRepo repository.ClusterRepository,
	maintenanceTaskRepo repository.MaintenanceTaskRepo,
	dedicatedCli dedicatedglobal.DedicatedGlobal,
	nextgenCli nextgenglobal.NextGenGlobal,
	billingCli billing.Client,
	accountCli account.Client,
	dbProvider db.TransProvider,
	cfg *config.Config,
	eventbus tieventbus.Eventbus, // Optional shared eventbus.
) Manager {
	if cfg.EventBus == nil {
		cfg.EventBus = &config.EventBus{Local: true}
	}

	m := &clusterEventConsumerManagerImp{
		jsmClient:           jsmClient,
		coreSvc:             coreClient,
		clusterRepo:         clusterRepo,
		maintenanceTaskRepo: maintenanceTaskRepo,
		dedicatedCli:        dedicatedCli,
		nextgenCli:          nextgenCli,
		billingCli:          billingCli,
		accountCli:          accountCli,
		uncConfig:           cfg.UserNotificationCenter,
		dbProvider:          dbProvider,
		customStatusRepo:    infrarepo.NewCustomStatusRepository(dbProvider),
	}

	// Set optional shared eventbus.
	if eventbus == nil {
		m.newEventbus(cfg.EventBus)
	} else {
		m.eventbus = eventbus
	}

	m.registerEventbusConsumers(cfg.EventBus)

	return m
}

func (c *clusterEventConsumerManagerImp) Run() error {
	return c.eventbus.Run()
}

func (c *clusterEventConsumerManagerImp) registerEventbusConsumers(cfg *config.EventBus) {
	// register 'sync_to_jsm_event' consumer
	syncToJsmHandler := tieventbus.HandlerConfig{
		EventType:       string(eventutil.TopicClusterEvent),
		Group:           "sync_to_jsm_event",
		NumOfGoroutines: cfg.NumOfGoroutines,
		RetryTimes:      cfg.RetryTimes,
		HandleFunc: eventbusConsumerHandleLogWrapper(func(ctx context.Context, ev *tieventbus.Event) error {
			return c.sendToJsmSqs(ctx, ev)
		}),
		DataStruct: eventutil.ClusterEvent{},
	}
	c.eventbus.RegisterHandler(syncToJsmHandler)
	prometheus.MustRegister(syncClusterInfoToJsmFailedTotal)

	// register 'cluster_display_event' consumer
	syncToDisplayEventHandler := tieventbus.HandlerConfig{
		EventType:       string(eventutil.TopicClusterEvent),
		Group:           "cluster_display_event",
		NumOfGoroutines: cfg.NumOfGoroutines,
		RetryTimes:      cfg.RetryTimes,
		HandleFunc: eventbusConsumerHandleLogWrapper(func(ctx context.Context, ev *tieventbus.Event) error {
			return c.sendToDisplayEvent(ctx, ev)
		}),
		DataStruct: eventutil.ClusterEvent{},
	}
	c.eventbus.RegisterHandler(syncToDisplayEventHandler)

	clusterMaintenanceDisplayEventHandler := tieventbus.HandlerConfig{
		EventType:       string(eventutil.TopicMaintenanceWindowEvent),
		Group:           "cluster_maintenance_display_event",
		NumOfGoroutines: cfg.NumOfGoroutines,
		RetryTimes:      cfg.RetryTimes,
		HandleFunc: eventbusConsumerHandleLogWrapper(func(ctx context.Context, ev *tieventbus.Event) error {
			return c.sendClusterMaintenanceDisplayEvent(ctx, ev)
		}),
		DataStruct: eventutil.MaintenanceWindowEvent{},
	}
	c.eventbus.RegisterHandler(clusterMaintenanceDisplayEventHandler)

	// register 'cluster_eventunc_event' consumer
	uncHandler := tieventbus.HandlerConfig{
		EventType:       string(eventutil.TopicClusterEvent),
		Group:           "user_notification_center",
		NumOfGoroutines: cfg.NumOfGoroutines,
		RetryTimes:      cfg.RetryTimes,
		HandleFunc: eventbusConsumerHandleLogWrapper(func(ctx context.Context, ev *tieventbus.Event) error {
			return c.sendToUserNotificationCenter(ctx, ev)
		}),
		DataStruct: eventutil.ClusterEvent{},
	}
	c.eventbus.RegisterHandler(uncHandler)

	// register 'cluster_event_email' consumer
	emailHandler := tieventbus.HandlerConfig{
		EventType:       string(eventutil.TopicClusterEvent),
		Group:           "cluster_event_email",
		NumOfGoroutines: cfg.NumOfGoroutines,
		RetryTimes:      cfg.RetryTimes,
		HandleFunc: eventbusConsumerHandleLogWrapper(func(ctx context.Context, ev *tieventbus.Event) error {
			return c.sendToClusterEventEmail(ctx, ev)
		}),
		DataStruct: eventutil.ClusterEvent{},
	}
	c.eventbus.RegisterHandler(emailHandler)
	prometheus.MustRegister(autoResumeClusterCountTotal)

	// register 'cluster status' consumer
	shadowPoolHandler := tieventbus.HandlerConfig{
		EventType:       string(eventutil.TopicClusterEvent),
		Group:           "shadow_pool",
		NumOfGoroutines: cfg.NumOfGoroutines,
		RetryTimes:      cfg.RetryTimes,
		HandleFunc: eventbusConsumerHandleLogWrapper(func(ctx context.Context, ev *tieventbus.Event) error {
			return c.handleShadowPoolInstanceState(ctx, ev)
		}),
		DataStruct: eventutil.ClusterEvent{},
	}
	c.eventbus.RegisterHandler(shadowPoolHandler)

	starterInventoryThresholdHandler := tieventbus.HandlerConfig{
		EventType:       string(eventutil.TopicClusterEvent),
		Group:           "starter_inventory_threshold",
		NumOfGoroutines: cfg.NumOfGoroutines,
		RetryTimes:      cfg.RetryTimes,
		HandleFunc: eventbusConsumerHandleLogWrapper(func(ctx context.Context, ev *tieventbus.Event) error {
			return c.checkStarterInventoryThresholdAndPublish(ctx, ev)
		}),
		DataStruct: eventutil.ClusterEvent{},
	}
	c.eventbus.RegisterHandler(starterInventoryThresholdHandler)

	rootPasswordRotatedHandler := tieventbus.HandlerConfig{
		EventType:       string(eventutil.TopicClusterEvent),
		Group:           "root_password_rotated_cleanup",
		NumOfGoroutines: cfg.NumOfGoroutines,
		RetryTimes:      cfg.RetryTimes,
		HandleFunc: eventbusConsumerHandleLogWrapper(func(ctx context.Context, ev *tieventbus.Event) error {
			return c.handleRootPasswordRotatedCleanup(ctx, ev)
		}),
		DataStruct: eventutil.ClusterEvent{},
	}
	c.eventbus.RegisterHandler(rootPasswordRotatedHandler)
}

func eventbusConsumerHandleLogWrapper(fn func(ctx context.Context, ev *tieventbus.Event) error) tieventbus.HandleFunc {
	return func(event *tieventbus.Event) error {
		ctx := context.Background()
		tilog.Info(ctx, "eventbus consumer handle event", zap.String("event_id", event.ID))

		if err := fn(ctx, event); err != nil {
			tilog.Error(ctx, "eventbus consumer handle event failed", zap.String("event_id", event.ID), zap.Error(err))
			return err
		}

		return nil
	}
}

func (c *clusterEventConsumerManagerImp) newEventbus(cfg *config.EventBus) tieventbus.Eventbus {
	if c.eventbus != nil {
		return c.eventbus
	}

	var saramaConfigModifier tieventbus.SaramaConfigModifier = func(conf *sarama.Config) {
		if cfg.ProxyAddress != "" {
			tilog.Info(context.Background(), "use eventbus proxy", zap.String("address", cfg.ProxyAddress))
			// enable proxy to support access tieventbus in your laptop(or local develop environment)
			proxyURL, err := url.Parse(cfg.ProxyAddress)
			if err != nil {
				panic(err)
			}
			p, err := proxy.FromURL(proxyURL, proxy.Direct)
			if err != nil {
				panic(err)
			}
			conf.Net.Proxy.Enable = true
			conf.Net.Proxy.Dialer = p
		}

		// pull messages from the newest point
		conf.Consumer.Offsets.Initial = sarama.OffsetNewest
	}
	eventConf := tieventbus.EventbusConfig{
		AppName:                         AppName,
		SaramaConfigModifier4Producer:   saramaConfigModifier,
		SaramaConfigModifier4Subscriber: saramaConfigModifier,
	}

	var err error

	if cfg.Local {
		c.eventbus, err = tieventbus.NewEventbusMock(eventConf)
	} else {
		c.eventbus, err = tieventbus.NewEventbus(eventConf)
	}
	if err != nil {
		panic(fmt.Sprintf("failed to initialize event bus, err=%s", err.Error()))
	}

	c.eventbus.EnableMetrics(prometheus.DefaultRegisterer, "", "")

	return c.eventbus
}