package auditloginit

import (
	"crypto/tls"
	"time"

	"github.com/Shopify/sarama"
	"github.com/tidbcloud/tiinf/transport/tieventbus"

	"github.com/pingcap-inc/clinic_cloud_server/pkg/config"
)

// NewEventbusConfig builds tieventbus options for the given app (e.g. auditlog / clinic).
// When audit_kafka.sasl.enable is false, modifiers are omitted (same as default SDK behavior).
func NewEventbusConfig(cfg *config.Config, appName string, mqTimeout time.Duration) tieventbus.EventbusConfig {
	eb := tieventbus.EventbusConfig{
		AppName:   appName,
		MQTimeout: mqTimeout,
	}
	if cfg == nil || !cfg.KafkaExt.SASL.Enabled {
		return eb
	}
	extConf := cfg.KafkaExt
	mod := func(conf *sarama.Config) {
		applySASL(conf, &extConf)
	}
	eb.SaramaConfigModifier4Producer = mod
	eb.SaramaConfigModifier4Subscriber = mod
	return eb
}

func applySASL(conf *sarama.Config, extConf *config.KafkaExtConfig) {
	useTLS := extConf.TLS.Enabled || extConf.SASL.Enabled
	if useTLS {
		conf.Net.TLS.Enable = true
		if extConf.TLS.InsecureSkipVerify {
			if conf.Net.TLS.Config == nil {
				conf.Net.TLS.Config = &tls.Config{
					InsecureSkipVerify: true,
					MinVersion:         tls.VersionTLS12,
				}
			} else {
				conf.Net.TLS.Config.InsecureSkipVerify = true
			}
		}
	}

	if extConf.SASL.Enabled {
		conf.Net.SASL.Enable = true
		conf.Net.SASL.User = extConf.SASL.Username
		conf.Net.SASL.Password = extConf.SASL.Password
		if extConf.SASL.Mechanism != "" {
			conf.Net.SASL.Mechanism = extConf.SASL.Mechanism
		} else {
			conf.Net.SASL.Mechanism = sarama.SASLTypePlaintext
		}
	}
}