package eventbus

import (
	"context"
	"errors"
	"fmt"

	"github.com/segmentio/kafka-go"

	"lingniu-vehicle-ingest/go/vehicle-gateway/internal/envelope"
)

type Sink interface {
	PublishRaw(context.Context, envelope.FrameEnvelope) error
	PublishUnified(context.Context, envelope.FrameEnvelope) error
	Close() error
}

type KafkaSink struct {
	writer       kafkaWriter
	rawTopics    map[envelope.Protocol]string
	unifiedTopic string
}

type kafkaWriter interface {
	WriteMessages(context.Context, ...kafka.Message) error
	Close() error
}

type KafkaConfig struct {
	Brokers      []string
	RawTopics    map[envelope.Protocol]string
	UnifiedTopic string
}

func NewKafkaSink(cfg KafkaConfig) (*KafkaSink, error) {
	if len(cfg.Brokers) == 0 {
		return nil, errors.New("kafka brokers are required")
	}
	return newKafkaSinkWithWriter(&kafka.Writer{
		Addr:                   kafka.TCP(cfg.Brokers...),
		Balancer:               &kafka.Hash{},
		AllowAutoTopicCreation: false,
	}, cfg), nil
}

func newKafkaSinkWithWriter(writer kafkaWriter, cfg KafkaConfig) *KafkaSink {
	rawTopics := map[envelope.Protocol]string{
		envelope.ProtocolGB32960:    "vehicle.raw.gb32960.v1",
		envelope.ProtocolJT808:      "vehicle.raw.jt808.v1",
		envelope.ProtocolYutongMQTT: "vehicle.raw.yutong-mqtt.v1",
	}
	for protocol, topic := range cfg.RawTopics {
		if topic != "" {
			rawTopics[protocol] = topic
		}
	}
	unifiedTopic := cfg.UnifiedTopic
	if unifiedTopic == "" {
		unifiedTopic = "vehicle.event.unified.v1"
	}
	return &KafkaSink{writer: writer, rawTopics: rawTopics, unifiedTopic: unifiedTopic}
}

func (s *KafkaSink) PublishRaw(ctx context.Context, env envelope.FrameEnvelope) error {
	topic, ok := s.rawTopics[env.Protocol]
	if !ok || topic == "" {
		return fmt.Errorf("raw topic not configured for protocol %s", env.Protocol)
	}
	return s.publish(ctx, topic, env)
}

func (s *KafkaSink) PublishUnified(ctx context.Context, env envelope.FrameEnvelope) error {
	return s.publish(ctx, s.unifiedTopic, env)
}

func (s *KafkaSink) Close() error {
	if s == nil || s.writer == nil {
		return nil
	}
	return s.writer.Close()
}

func (s *KafkaSink) publish(ctx context.Context, topic string, env envelope.FrameEnvelope) error {
	payload, err := env.MarshalJSONBytes()
	if err != nil {
		return err
	}
	return s.writer.WriteMessages(ctx, kafka.Message{
		Topic: topic,
		Key:   env.KafkaKey(),
		Value: payload,
	})
}