package eventbus

import (
	"crypto/tls"
	"fmt"

	"github.com/tidbcloud/devops/entity/event"
	"github.com/tidbcloud/devops/pkg/infra/config"

	"github.com/Shopify/sarama"
	"github.com/go-logr/logr"
	"github.com/google/uuid"
	evbus "github.com/tidbcloud/eventbus-datastruct/event"
	mgmtevent "github.com/tidbcloud/tidb-management-service/utilities/clusterevent"
	"github.com/tidbcloud/tiinf/transport/tieventbus"
	"go.uber.org/dig"
)

const (
	topicChangeEvent                 = "devops_change_event"
	topicClusterMaintenanceEvent     = "devops_cluster_maintenance_event"
	topicOperationEvent              = "devops_operation_event"
	topicJumpEvent                   = "devops_jump_event"
	topicApprovalEvent               = "devops_approval_event"
	topicK8sCalendarEvent            = "devops_k8s_calendar_event"
	topicDBaasMaintenanceWindowEvent = string(mgmtevent.TopicMaintenanceWindowEvent)
	topicDBaasTenantPlanEvent        = "tenant_plan"
	topicDBaasTenantEvent            = "OrgEvent"
	topicDBaasClusterEvent           = string(mgmtevent.TopicClusterEvent)
	topicDisplayEvent                = "display_event"
)

type EventBusParam struct {
	dig.In

	Conf   *config.RootConfig
	Logger logr.Logger
}

type EventBus struct {
	logger logr.Logger
	bus    tieventbus.Eventbus

	// internal
	ChangeEvent             *pubsub[event.ChangeEvent]
	ClusterMaintenanceEvent *pubsub[event.ClusterMaintenanceEvent]
	OperationEvent          *pubsub[event.OperationEvent]
	JumpEvent               *pubsub[event.JumpEvent]
	ApprovalEvent           *pubsub[event.ApprovalEvent]
	K8sCalendarEvent        *pubsub[event.K8sCalendarEvent]
	// external
	DBaasMaintenanceWindowEvent *pubsub[mgmtevent.MaintenanceWindowEvent]
	DBaasTenantEvent            *pubsub[event.TenantEvent]
	DBaasTenantPlanEvent        *pubsub[event.TenantPlanEvent]
	DBaaSClusterEvent           *pubsub[mgmtevent.ClusterEvent]
	DBaaSDisplayEvent           *pubsub[evbus.DisplayEvent]
}

func NewEventBus(p EventBusParam) (*EventBus, error) {
	var err error
	b := &EventBus{
		logger: p.Logger.WithName("EventBus"),
	}

	conf := p.Conf.EventBus

	if conf.Disable {
		b.bus = &DiscardEventBus{}
	} else {
		saramaConfigMutate := func(saramaConfig *sarama.Config) {
			if conf.AuthSASL != nil {
				saramaConfig.Net.TLS.Enable = true
				saramaConfig.Net.TLS.Config = &tls.Config{
					InsecureSkipVerify: true, //nolint:gosec
				}
				saramaConfig.Net.SASL.Enable = true
				saramaConfig.Net.SASL.User = conf.AuthSASL.Username
				saramaConfig.Net.SASL.Password = conf.AuthSASL.Password
				saramaConfig.Net.SASL.Mechanism = sarama.SASLTypePlaintext
				saramaConfig.Producer.Return.Successes = true
				saramaConfig.Version = sarama.V2_0_0_0
			}
		}

		b.bus, err = tieventbus.NewEventbus(tieventbus.EventbusConfig{
			AppName: conf.AppName,

			SaramaConfigModifier4Producer:   saramaConfigMutate,
			SaramaConfigModifier4Subscriber: saramaConfigMutate,
		})
		if err != nil {
			return nil, err
		}
	}

	b.ChangeEvent = newPubSub[event.ChangeEvent](topicChangeEvent, b.logger, b.bus)
	b.ClusterMaintenanceEvent = newPubSub[event.ClusterMaintenanceEvent](topicClusterMaintenanceEvent, b.logger, b.bus)
	b.OperationEvent = newPubSub[event.OperationEvent](topicOperationEvent, b.logger, b.bus)
	b.JumpEvent = newPubSub[event.JumpEvent](topicJumpEvent, b.logger, b.bus)
	b.ApprovalEvent = newPubSub[event.ApprovalEvent](topicApprovalEvent, b.logger, b.bus)
	b.K8sCalendarEvent = newPubSub[event.K8sCalendarEvent](topicK8sCalendarEvent, b.logger, b.bus)

	b.DBaasMaintenanceWindowEvent = newPubSub[mgmtevent.MaintenanceWindowEvent](topicDBaasMaintenanceWindowEvent, b.logger, b.bus)
	b.DBaasTenantEvent = newPubSub[event.TenantEvent](topicDBaasTenantEvent, b.logger, b.bus)
	b.DBaasTenantPlanEvent = newPubSub[event.TenantPlanEvent](topicDBaasTenantPlanEvent, b.logger, b.bus)
	b.DBaaSClusterEvent = newPubSub[mgmtevent.ClusterEvent](topicDBaasClusterEvent, b.logger, b.bus)
	b.DBaaSDisplayEvent = newPubSub[evbus.DisplayEvent](topicDisplayEvent, b.logger, b.bus)

	return b, nil
}

func NewMockEventBus() *EventBus {
	param := EventBusParam{
		Conf:   &config.RootConfig{},
		Logger: logr.Discard(),
	}
	param.Conf.EventBus.Disable = true

	eventbus, err := NewEventBus(param)
	if err != nil {
		panic(err)
	}

	return eventbus
}

func (b *EventBus) Run() error {
	if err := b.ChangeEvent.run(); err != nil {
		return err
	}
	if err := b.ClusterMaintenanceEvent.run(); err != nil {
		return err
	}
	if err := b.OperationEvent.run(); err != nil {
		return err
	}
	if err := b.JumpEvent.run(); err != nil {
		return err
	}
	if err := b.ApprovalEvent.run(); err != nil {
		return err
	}
	if err := b.K8sCalendarEvent.run(); err != nil {
		return err
	}

	if err := b.DBaasMaintenanceWindowEvent.run(); err != nil {
		return err
	}
	if err := b.DBaasTenantEvent.run(); err != nil {
		return err
	}
	if err := b.DBaasTenantPlanEvent.run(); err != nil {
		return err
	}
	if err := b.DBaaSClusterEvent.run(); err != nil {
		return err
	}
	if err := b.DBaaSDisplayEvent.run(); err != nil {
		return err
	}

	return b.bus.Run()
}

func (b *EventBus) Close() {
	b.bus.Close()
}

func genUUID(typ string) string {
	return fmt.Sprintf("%s-%s", typ, uuid.New().String())
}