package main

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

	"github.com/segmentio/kafka-go"
	"github.com/tinode/chat/server/logs"
)

const (
	kafkaTopicAdminCommand       = "im.admin.command"
	kafkaTopicAdminCommandResult = "im.admin.command.result"
	kafkaTopicConfigEvent        = "im.config.event"
	kafkaTopicBusinessEvent      = "im.business.event"
	kafkaTopicMessagePersist     = "im.message.persist"
	kafkaTopicMessagePush        = "im.message.push"
	kafkaTopicFileSecurity       = "im.file.security"
	kafkaTopicFileLifecycle      = "im.file.lifecycle"
	kafkaTopicWorkbenchEvent     = "im.workbench.event"
	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"`
	RequiredAcks     int               `json:"required_acks"`
	AllowAutoTopic   bool              `json:"allow_auto_topic_creation"`
	TLSEnabled       bool              `json:"tls_enabled"`
	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"`
	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")
	}

	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}
	}
	rt.writer = &kafka.Writer{
		Addr:                   kafka.TCP(cfg.Brokers...),
		Balancer:               &kafka.Hash{},
		Transport:              transport,
		RequiredAcks:           kafka.RequiredAcks(cfg.RequiredAcks),
		AllowAutoTopicCreation: cfg.AllowAutoTopic,
		WriteTimeout:           time.Duration(cfg.WriteTimeoutMS) * time.Millisecond,
		ReadTimeout:            time.Duration(cfg.ReadTimeoutMS) * time.Millisecond,
	}

	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.RequiredAcks == 0 {
		cfg.RequiredAcks = int(kafka.RequireOne)
	}
	if cfg.MaxRetryCount <= 0 {
		cfg.MaxRetryCount = 5
	}
	if len(cfg.RetryBackoffMS) == 0 {
		cfg.RetryBackoffMS = []int{10000, 30000, 60000, 300000, 900000}
	}
	if cfg.DeadRetentionSec <= 0 {
		cfg.DeadRetentionSec = 90 * 24 * 3600
	}
}

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,
		"message_push":         kafkaTopicMessagePush,
		"file_security":        kafkaTopicFileSecurity,
		"file_lifecycle":       kafkaTopicFileLifecycle,
		"workbench_event":      kafkaTopicWorkbenchEvent,
		"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
	}
	reader := kafka.NewReader(kafka.ReaderConfig{
		Brokers:     kafkaBus.config.Brokers,
		GroupID:     kafkaBus.config.ConsumerGroup,
		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, kafkaBus.config.ConsumerGroup)
}

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}
	}
	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
}