package main

import (
	"context"
	"crypto/tls"
	"encoding/json"
	"errors"
	"strings"
	"sync"
	"time"

	"github.com/segmentio/kafka-go"
	"github.com/segmentio/kafka-go/sasl"
	"github.com/segmentio/kafka-go/sasl/scram"
	"github.com/tinode/chat/server/logs"
	"github.com/tinode/chat/server/store/types"
)

const (
	kafkaTopicAdminCommand        = "im.admin.command"
	kafkaTopicAdminCommandResult  = "im.admin.command.result"
	kafkaTopicConfigEvent         = "im.config.event"
	kafkaTopicBusinessEvent       = "im.business.event"
	kafkaTopicMessagePersist      = "im.message.persist"
	kafkaTopicGroupMessagePersist = "im.group.message.persist"
	kafkaTopicMessagePush         = "im.message.push"
	kafkaTopicFileSecurity        = "im.file.security"
	kafkaTopicCacheInvalidate     = "im.cache.invalidate"
	kafkaTopicAuditEvent          = "im.audit.event"
	kafkaTopicDeadLetter          = "im.dead.letter"
)

type kafkaConfig struct {
	Enabled          bool              `json:"enabled"`
	Brokers          []string          `json:"brokers"`
	ClientID         string            `json:"client_id"`
	ConsumerGroup    string            `json:"consumer_group"`
	Topics           map[string]string `json:"topics"`
	DialTimeoutMS    int               `json:"dial_timeout_ms"`
	WriteTimeoutMS   int               `json:"write_timeout_ms"`
	ReadTimeoutMS    int               `json:"read_timeout_ms"`
	BatchTimeoutMS   int               `json:"batch_timeout_ms"`
	RequiredAcks     int               `json:"required_acks"`
	AllowAutoTopic   bool              `json:"allow_auto_topic_creation"`
	TLSEnabled       bool              `json:"tls_enabled"`
	SASLEnabled      bool              `json:"sasl_enabled"`
	SASLMechanism    string            `json:"sasl_mechanism"`
	SASLUsername     string            `json:"sasl_username"`
	SASLPassword     string            `json:"sasl_password"`
	MaxRetryCount    int               `json:"max_retry_count"`
	RetryBackoffMS   []int             `json:"retry_backoff_ms"`
	DeadRetentionSec int               `json:"dead_letter_retention_seconds"`
}

type kafkaEnvelope struct {
	EventID        string          `json:"event_id"`
	TraceID        string          `json:"trace_id,omitempty"`
	TenantCode     string          `json:"tenant_code,omitempty"`
	TenantID       types.TenantID  `json:"tenant_id,omitempty"`
	SourceService  string          `json:"source_service,omitempty"`
	TargetService  string          `json:"target_service,omitempty"`
	EventType      string          `json:"event_type"`
	BizType        string          `json:"biz_type,omitempty"`
	BizID          string          `json:"biz_id,omitempty"`
	IdempotencyKey string          `json:"idempotency_key,omitempty"`
	Version        int64           `json:"version,omitempty"`
	OccurredAt     time.Time       `json:"occurred_at"`
	Actor          *kafkaActor     `json:"actor,omitempty"`
	Payload        json.RawMessage `json:"payload,omitempty"`
	ParentEventID  string          `json:"parent_event_id,omitempty"`
}

type kafkaActor struct {
	Type        string `json:"type,omitempty"`
	ID          string `json:"id,omitempty"`
	DisplayName string `json:"display_name,omitempty"`
}

type kafkaRuntime struct {
	config kafkaConfig
	topics map[string]string

	writer *kafka.Writer
	ctx    context.Context
	cancel context.CancelFunc
	wg     sync.WaitGroup
}

var kafkaBus *kafkaRuntime

func initKafkaRuntime(raw json.RawMessage) error {
	if len(raw) == 0 {
		kafkaBus = nil
		logs.Info.Println("Kafka disabled: no config")
		return nil
	}
	var cfg kafkaConfig
	if err := json.Unmarshal(raw, &cfg); err != nil {
		return err
	}
	if !cfg.Enabled {
		kafkaBus = nil
		logs.Info.Println("Kafka disabled")
		return nil
	}
	normalizeKafkaConfig(&cfg)
	if len(cfg.Brokers) == 0 {
		return errors.New("kafka: enabled but brokers are not configured")
	}
	saslMechanism, err := kafkaSASLMechanism(cfg)
	if err != nil {
		return err
	}

	ctx, cancel := context.WithCancel(context.Background())
	rt := &kafkaRuntime{
		config: cfg,
		topics: kafkaTopics(cfg.Topics),
		ctx:    ctx,
		cancel: cancel,
	}
	transport := &kafka.Transport{
		ClientID: cfg.ClientID,
	}
	if cfg.TLSEnabled {
		transport.TLS = &tls.Config{MinVersion: tls.VersionTLS12}
	}
	transport.SASL = saslMechanism
	rt.writer = &kafka.Writer{
		Addr:                   kafka.TCP(cfg.Brokers...),
		Balancer:               &kafka.Hash{},
		Transport:              transport,
		RequiredAcks:           kafka.RequiredAcks(cfg.RequiredAcks),
		AllowAutoTopicCreation: cfg.AllowAutoTopic,
		BatchTimeout:           time.Duration(cfg.BatchTimeoutMS) * time.Millisecond,
		WriteTimeout:           time.Duration(cfg.WriteTimeoutMS) * time.Millisecond,
		ReadTimeout:            time.Duration(cfg.ReadTimeoutMS) * time.Millisecond,
		MaxAttempts:            cfg.MaxRetryCount,
	}

	if err := rt.ping(context.Background()); err != nil {
		cancel()
		_ = rt.writer.Close()
		return err
	}
	kafkaBus = rt
	logs.Info.Printf("Kafka enabled: brokers=%s client_id=%s group=%s", strings.Join(cfg.Brokers, ","), cfg.ClientID, cfg.ConsumerGroup)
	return nil
}

func normalizeKafkaConfig(cfg *kafkaConfig) {
	if cfg.ClientID == "" {
		cfg.ClientID = "im-server"
	}
	if cfg.ConsumerGroup == "" {
		cfg.ConsumerGroup = "im-server"
	}
	if cfg.DialTimeoutMS <= 0 {
		cfg.DialTimeoutMS = 3000
	}
	if cfg.WriteTimeoutMS <= 0 {
		cfg.WriteTimeoutMS = 3000
	}
	if cfg.ReadTimeoutMS <= 0 {
		cfg.ReadTimeoutMS = 3000
	}
	if cfg.BatchTimeoutMS <= 0 {
		cfg.BatchTimeoutMS = 10
	}
	if cfg.RequiredAcks == 0 {
		cfg.RequiredAcks = int(kafka.RequireOne)
	}
	if cfg.MaxRetryCount <= 0 {
		cfg.MaxRetryCount = 3
	}
	if len(cfg.RetryBackoffMS) == 0 {
		cfg.RetryBackoffMS = []int{10000, 30000, 60000, 300000, 900000}
	}
	if cfg.DeadRetentionSec <= 0 {
		cfg.DeadRetentionSec = 90 * 24 * 3600
	}
	if cfg.SASLMechanism == "" {
		cfg.SASLMechanism = "SCRAM-SHA-512"
	}
}

func kafkaSASLMechanism(cfg kafkaConfig) (sasl.Mechanism, error) {
	if !cfg.SASLEnabled && cfg.SASLUsername == "" {
		return nil, nil
	}
	if cfg.SASLUsername == "" || cfg.SASLPassword == "" {
		return nil, errors.New("kafka: sasl enabled but username or password is empty")
	}
	switch strings.ToUpper(strings.ReplaceAll(cfg.SASLMechanism, "_", "-")) {
	case "", "SCRAM-SHA-512":
		return scram.Mechanism(scram.SHA512, cfg.SASLUsername, cfg.SASLPassword)
	case "SCRAM-SHA-256":
		return scram.Mechanism(scram.SHA256, cfg.SASLUsername, cfg.SASLPassword)
	default:
		return nil, errors.New("kafka: unsupported sasl mechanism " + cfg.SASLMechanism)
	}
}

func kafkaTopics(overrides map[string]string) map[string]string {
	topics := map[string]string{
		"admin_command":         kafkaTopicAdminCommand,
		"admin_command_result":  kafkaTopicAdminCommandResult,
		"config_event":          kafkaTopicConfigEvent,
		"business_event":        kafkaTopicBusinessEvent,
		"message_persist":       kafkaTopicMessagePersist,
		"group_message_persist": kafkaTopicGroupMessagePersist,
		"message_push":          kafkaTopicMessagePush,
		"file_security":         kafkaTopicFileSecurity,
		"cache_invalidate":      kafkaTopicCacheInvalidate,
		"audit_event":           kafkaTopicAuditEvent,
		"dead_letter":           kafkaTopicDeadLetter,
	}
	for key, value := range overrides {
		if value != "" {
			topics[key] = value
		}
	}
	return topics
}

func (rt *kafkaRuntime) ping(ctx context.Context) error {
	timeout := time.Duration(rt.config.DialTimeoutMS) * time.Millisecond
	dialer := rt.dialer()
	pingCtx, cancel := context.WithTimeout(ctx, timeout)
	defer cancel()
	conn, err := dialer.DialContext(pingCtx, "tcp", rt.config.Brokers[0])
	if err != nil {
		return err
	}
	return conn.Close()
}

func kafkaEnabled() bool {
	return kafkaBus != nil
}

func kafkaTopic(name string) string {
	if kafkaBus == nil {
		return ""
	}
	return kafkaBus.topics[name]
}

func publishKafkaEnvelope(ctx context.Context, topicName string, key string, envelope kafkaEnvelope) error {
	if kafkaBus == nil {
		return nil
	}
	if envelope.OccurredAt.IsZero() {
		envelope.OccurredAt = time.Now().UTC()
	}
	topic := kafkaBus.topics[topicName]
	if topic == "" {
		topic = topicName
	}
	value, err := json.Marshal(envelope)
	if err != nil {
		return err
	}
	return kafkaBus.writer.WriteMessages(ctx, kafka.Message{
		Topic: topic,
		Key:   []byte(key),
		Value: value,
		Time:  envelope.OccurredAt,
	})
}

type kafkaMessageHandler func(context.Context, kafkaEnvelope, kafka.Message) error

func startKafkaConsumer(topicName string, handler kafkaMessageHandler) {
	if kafkaBus == nil || handler == nil {
		return
	}
	topic := kafkaBus.topics[topicName]
	if topic == "" {
		topic = topicName
	}
	groupID := kafkaBus.config.ConsumerGroup
	if topicName != "" {
		groupID += "." + topicName
	}
	reader := kafka.NewReader(kafka.ReaderConfig{
		Brokers:     kafkaBus.config.Brokers,
		GroupID:     groupID,
		Topic:       topic,
		StartOffset: kafka.FirstOffset,
		MinBytes:    1,
		MaxBytes:    10e6,
		MaxWait:     time.Second,
		Dialer:      kafkaBus.dialer(),
	})
	kafkaBus.wg.Add(1)
	go func() {
		defer kafkaBus.wg.Done()
		defer reader.Close()
		for {
			msg, err := reader.FetchMessage(kafkaBus.ctx)
			if err != nil {
				if errors.Is(err, context.Canceled) || errors.Is(kafkaBus.ctx.Err(), context.Canceled) {
					return
				}
				logs.Err.Println("Kafka fetch failed:", err)
				continue
			}
			var envelope kafkaEnvelope
			if err = json.Unmarshal(msg.Value, &envelope); err != nil {
				logs.Err.Println("Kafka envelope decode failed:", err)
				_ = reader.CommitMessages(context.Background(), msg)
				continue
			}
			if err = handler(kafkaBus.ctx, envelope, msg); err != nil {
				logs.Err.Println("Kafka handler failed:", envelope.EventID, envelope.EventType, err)
				continue
			}
			if err = reader.CommitMessages(context.Background(), msg); err != nil {
				logs.Err.Println("Kafka commit failed:", err)
			}
		}
	}()
	logs.Info.Printf("Kafka consumer started: topic=%s group=%s", topic, groupID)
}

func (rt *kafkaRuntime) dialer() *kafka.Dialer {
	dialer := &kafka.Dialer{
		ClientID: rt.config.ClientID,
		Timeout:  time.Duration(rt.config.DialTimeoutMS) * time.Millisecond,
	}
	if rt.config.TLSEnabled {
		dialer.TLS = &tls.Config{MinVersion: tls.VersionTLS12}
	}
	if mechanism, err := kafkaSASLMechanism(rt.config); err != nil {
		logs.Err.Println("Kafka SASL config failed:", err)
	} else {
		dialer.SASLMechanism = mechanism
	}
	return dialer
}

func stopKafkaRuntime() {
	if kafkaBus == nil {
		return
	}
	kafkaBus.cancel()
	kafkaBus.wg.Wait()
	if kafkaBus.writer != nil {
		_ = kafkaBus.writer.Close()
	}
	logs.Info.Println("Stopped Kafka runtime")
	kafkaBus = nil
}