package main

import (
	"context"
	"encoding/json"
	"errors"
	"fmt"
	"log/slog"
	"os"
	"os/signal"
	"sort"
	"strconv"
	"strings"
	"syscall"
	"time"

	"github.com/nats-io/nats.go"
	"github.com/segmentio/kafka-go"

	"lingniu-vehicle-ingest/go/vehicle-gateway/internal/envelope"
	"lingniu-vehicle-ingest/go/vehicle-gateway/internal/health"
	"lingniu-vehicle-ingest/go/vehicle-gateway/internal/metrics"
	"lingniu-vehicle-ingest/go/vehicle-gateway/internal/observability"
	"lingniu-vehicle-ingest/go/vehicle-gateway/internal/realtime"
	"lingniu-vehicle-ingest/go/vehicle-gateway/internal/topics"
)

func main() {
	logger := observability.NewLogger("nats-kafka-bridge")
	ctx, stop := signal.NotifyContext(context.Background(), os.Interrupt, syscall.SIGTERM)
	defer stop()

	cfg := loadConfig()
	if err := cfg.Validate(); err != nil {
		logger.Error("invalid bridge config", "error", err)
		os.Exit(1)
	}
	conn, err := nats.Connect(cfg.NATSURL, nats.Name(cfg.NATSClientName), nats.Timeout(5*time.Second))
	if err != nil {
		logger.Error("nats connect failed", "error", err)
		os.Exit(1)
	}
	defer conn.Close()
	registry := metrics.NewRegistry()
	recordBridgeConfigMetrics(registry, cfg)
	health.Start(ctx, logger, health.NewServer(env("HEALTH_ADDR", ""), "nats-kafka-bridge", []health.Check{
		{Name: "nats", Check: func(context.Context) error {
			if conn.Status() != nats.CONNECTED {
				return fmt.Errorf("nats status is %s", conn.Status().String())
			}
			return nil
		}},
	}, registry))
	js, err := conn.JetStream()
	if err != nil {
		logger.Error("nats jetstream init failed", "error", err)
		os.Exit(1)
	}
	if err := ensureStream(js, cfg); err != nil {
		logger.Error("nats stream ensure failed", "stream", cfg.NATSStream, "error", err)
		os.Exit(1)
	}
	sub, err := js.PullSubscribe(
		cfg.NATSFilter,
		cfg.NATSDurable,
		nats.BindStream(cfg.NATSStream),
		nats.ManualAck(),
		nats.AckWait(cfg.AckWait),
		nats.MaxDeliver(-1),
	)
	if err != nil {
		logger.Error("nats pull consumer init failed", "stream", cfg.NATSStream, "durable", cfg.NATSDurable, "error", err)
		os.Exit(1)
	}

	logger.Info("nats kafka bridge started",
		"nats_url", cfg.NATSURL,
		"stream", cfg.NATSStream,
		"durable", cfg.NATSDurable,
		"filter", cfg.NATSFilter,
		"kafka_brokers", strings.Join(cfg.KafkaBrokers, ","),
		"kafka_batch_timeout_ms", cfg.KafkaBatchTimeout.Milliseconds(),
		"kafka_write_concurrency", cfg.KafkaWriteConcurrency,
		"fetch_wait_ms", cfg.FetchWait.Milliseconds(),
		"batch_size", cfg.BatchSize,
		"workers", cfg.Workers,
		"derive_fields_from_raw_enabled", cfg.DeriveFieldsFromRaw)
	for i := 0; i < cfg.Workers; i++ {
		writer := newKafkaWriter(cfg)
		defer writer.Close()
		go runBridge(ctx, logger.With("worker", i), registry, js, sub, writer, cfg)
	}
	<-ctx.Done()
}

type config struct {
	NATSURL               string
	NATSClientName        string
	NATSStream            string
	NATSDurable           string
	NATSFilter            string
	NATSSubjects          []string
	KafkaBrokers          []string
	RawRoutes             []subjectRoute
	FieldsRoutes          []subjectRoute
	Route                 map[string]string
	RawFieldRoutes        map[string]fieldsProjectionRoute
	DeriveFieldsFromRaw   bool
	KafkaBatchTimeout     time.Duration
	KafkaWriteConcurrency int
	BatchSize             int
	FetchWait             time.Duration
	OperationWait         time.Duration
	AckWait               time.Duration
	StreamMaxAge          time.Duration
	StreamMaxBytes        int64
	StreamEnsureWait      time.Duration
	Workers               int
}

type subjectRoute struct {
	Protocol envelope.Protocol
	Subject  string
	Topic    string
}

type fieldsProjectionRoute struct {
	Protocol envelope.Protocol
	Topic    string
}

func loadConfig() config {
	rawRoutes := []subjectRoute{
		{Protocol: envelope.ProtocolGB32960, Subject: env("NATS_SUBJECT_GB32960_RAW", env("KAFKA_TOPIC_GB32960_RAW", topics.RawGB32960)), Topic: env("KAFKA_TOPIC_GB32960_RAW", topics.RawGB32960)},
		{Protocol: envelope.ProtocolJT808, Subject: env("NATS_SUBJECT_JT808_RAW", env("KAFKA_TOPIC_JT808_RAW", topics.RawJT808)), Topic: env("KAFKA_TOPIC_JT808_RAW", topics.RawJT808)},
		{Protocol: envelope.ProtocolYutongMQTT, Subject: env("NATS_SUBJECT_YUTONG_MQTT_RAW", env("KAFKA_TOPIC_YUTONG_MQTT_RAW", topics.RawYutongMQTT)), Topic: env("KAFKA_TOPIC_YUTONG_MQTT_RAW", topics.RawYutongMQTT)},
	}
	fieldsRoutes := []subjectRoute{
		{Protocol: envelope.ProtocolGB32960, Subject: env("NATS_SUBJECT_GB32960_FIELDS", env("KAFKA_TOPIC_GB32960_FIELDS", topics.FieldsGB32960)), Topic: env("KAFKA_TOPIC_GB32960_FIELDS", topics.FieldsGB32960)},
		{Protocol: envelope.ProtocolJT808, Subject: env("NATS_SUBJECT_JT808_FIELDS", env("KAFKA_TOPIC_JT808_FIELDS", topics.FieldsJT808)), Topic: env("KAFKA_TOPIC_JT808_FIELDS", topics.FieldsJT808)},
		{Protocol: envelope.ProtocolYutongMQTT, Subject: env("NATS_SUBJECT_YUTONG_MQTT_FIELDS", env("KAFKA_TOPIC_YUTONG_MQTT_FIELDS", topics.FieldsYutongMQTT)), Topic: env("KAFKA_TOPIC_YUTONG_MQTT_FIELDS", topics.FieldsYutongMQTT)},
	}
	route := routeMap(rawRoutes, fieldsRoutes)
	if unifiedSubject, unifiedTopic, ok := unifiedRouteFromEnv(); ok {
		route[unifiedSubject] = unifiedTopic
	}
	workers := envInt("BRIDGE_WORKERS", 4)
	if workers < 1 {
		workers = 1
	}
	return config{
		NATSURL:               env("NATS_URL", "nats://127.0.0.1:4222"),
		NATSClientName:        env("NATS_CLIENT_NAME", "lingniu-nats-kafka-bridge"),
		NATSStream:            env("NATS_STREAM", "VEHICLE_INGEST"),
		NATSDurable:           env("NATS_DURABLE", "vehicle-kafka-bridge"),
		NATSFilter:            env("NATS_FILTER", "vehicle.>"),
		NATSSubjects:          splitCSV(env("NATS_STREAM_SUBJECTS", strings.Join(mapKeys(route), ","))),
		KafkaBrokers:          splitCSV(env("KAFKA_BROKERS", "127.0.0.1:9092")),
		RawRoutes:             rawRoutes,
		FieldsRoutes:          fieldsRoutes,
		Route:                 route,
		RawFieldRoutes:        rawFieldProjectionRoutes(rawRoutes, fieldsRoutes),
		DeriveFieldsFromRaw:   envBool("BRIDGE_DERIVE_FIELDS_FROM_RAW_ENABLED", true),
		KafkaBatchTimeout:     time.Duration(envInt("BRIDGE_KAFKA_BATCH_TIMEOUT_MS", 20)) * time.Millisecond,
		KafkaWriteConcurrency: envInt("BRIDGE_KAFKA_WRITE_CONCURRENCY", 6),
		BatchSize:             envInt("BRIDGE_BATCH_SIZE", 500),
		FetchWait:             time.Duration(envInt("BRIDGE_FETCH_WAIT_MS", 20)) * time.Millisecond,
		OperationWait:         time.Duration(envInt("BRIDGE_OPERATION_TIMEOUT_MS", 30000)) * time.Millisecond,
		AckWait:               time.Duration(envInt("NATS_ACK_WAIT_SECONDS", 60)) * time.Second,
		StreamMaxAge:          time.Duration(envInt("NATS_STREAM_MAX_AGE_HOURS", 24)) * time.Hour,
		StreamMaxBytes:        envInt64("NATS_STREAM_MAX_BYTES", 20*1024*1024*1024),
		StreamEnsureWait:      time.Duration(envInt("NATS_STREAM_ENSURE_TIMEOUT_SECONDS", 60)) * time.Second,
		Workers:               workers,
	}
}

func recordBridgeConfigMetrics(registry *metrics.Registry, cfg config) {
	if registry == nil {
		return
	}
	registry.SetGauge("vehicle_bridge_config", metrics.Labels{"setting": "workers"}, float64(cfg.Workers))
	registry.SetGauge("vehicle_bridge_config", metrics.Labels{"setting": "batch_size"}, float64(cfg.BatchSize))
	registry.SetGauge("vehicle_bridge_config", metrics.Labels{"setting": "fetch_wait_ms"}, float64(cfg.FetchWait.Milliseconds()))
	registry.SetGauge("vehicle_bridge_config", metrics.Labels{"setting": "kafka_batch_timeout_ms"}, float64(cfg.KafkaBatchTimeout.Milliseconds()))
	registry.SetGauge("vehicle_bridge_config", metrics.Labels{"setting": "kafka_write_concurrency"}, float64(cfg.KafkaWriteConcurrency))
	registry.SetGauge("vehicle_bridge_config", metrics.Labels{"setting": "derive_fields_from_raw_enabled"}, boolMetric(cfg.DeriveFieldsFromRaw))
}

func newKafkaWriter(cfg config) *kafka.Writer {
	return &kafka.Writer{
		Addr:                   kafka.TCP(cfg.KafkaBrokers...),
		Balancer:               &kafka.Hash{},
		AllowAutoTopicCreation: false,
		RequiredAcks:           kafka.RequireAll,
		BatchTimeout:           cfg.KafkaBatchTimeout,
		Async:                  false,
	}
}

func routeMap(rawRoutes []subjectRoute, fieldsRoutes []subjectRoute) map[string]string {
	route := make(map[string]string, len(rawRoutes)+len(fieldsRoutes))
	for _, item := range rawRoutes {
		if item.Subject != "" {
			route[item.Subject] = item.Topic
		}
	}
	for _, item := range fieldsRoutes {
		if item.Subject != "" {
			route[item.Subject] = item.Topic
		}
	}
	return route
}

func rawFieldProjectionRoutes(rawRoutes []subjectRoute, fieldsRoutes []subjectRoute) map[string]fieldsProjectionRoute {
	fieldsByProtocol := make(map[envelope.Protocol]string, len(fieldsRoutes))
	for _, route := range fieldsRoutes {
		if route.Protocol != "" && strings.TrimSpace(route.Topic) != "" {
			fieldsByProtocol[route.Protocol] = strings.TrimSpace(route.Topic)
		}
	}
	out := make(map[string]fieldsProjectionRoute, len(rawRoutes))
	for _, route := range rawRoutes {
		subject := strings.TrimSpace(route.Subject)
		topic := fieldsByProtocol[route.Protocol]
		if subject == "" || topic == "" {
			continue
		}
		out[subject] = fieldsProjectionRoute{Protocol: route.Protocol, Topic: topic}
	}
	return out
}

func (c config) Validate() error {
	rawSubjects, rawTopics := routeSubjectsAndTopics(c.RawRoutes)
	fieldsSubjects, fieldsTopics := routeSubjectsAndTopics(c.FieldsRoutes)
	if err := topics.ValidateKafkaRawFields(rawTopics, fieldsTopics); err != nil {
		return err
	}
	return topics.ValidateRawFieldsDisjoint(rawSubjects, fieldsSubjects, "nats subject")
}

func routeSubjectsAndTopics(routes []subjectRoute) (map[string]string, map[string]string) {
	subjects := make(map[string]string, len(routes))
	kafkaTopics := make(map[string]string, len(routes))
	for _, route := range routes {
		name := strings.TrimSpace(route.Subject)
		if name == "" {
			name = strings.TrimSpace(route.Topic)
		}
		if name == "" {
			name = "unknown"
		}
		subjects[name] = strings.TrimSpace(route.Subject)
		topicName := routeProtocolName(route)
		if topicName == "" {
			topicName = name
		}
		kafkaTopics[topicName] = strings.TrimSpace(route.Topic)
	}
	return subjects, kafkaTopics
}

func routeProtocolName(route subjectRoute) string {
	if protocol := strings.TrimSpace(string(route.Protocol)); protocol != "" {
		return protocol
	}
	if protocol, ok := topics.ProtocolForKnownRawTopic(route.Topic); ok {
		return protocol
	}
	if protocol, ok := topics.ProtocolForKnownFieldsTopic(route.Topic); ok {
		return protocol
	}
	return ""
}

func unifiedRouteFromEnv() (string, string, bool) {
	subject, hasSubject := envOptional("NATS_SUBJECT_UNIFIED")
	topic, hasTopic := envOptional("KAFKA_TOPIC_UNIFIED")
	if !hasSubject && !hasTopic {
		return "", "", false
	}
	if subject == "" {
		subject = topic
	}
	if topic == "" {
		topic = subject
	}
	if subject == "" || topic == "" {
		return "", "", false
	}
	return subject, topic, true
}

type natsPullSubscription interface {
	Fetch(int, ...nats.PullOpt) ([]*nats.Msg, error)
}

type natsConsumerInfoReader interface {
	ConsumerInfo(stream string, name string, opts ...nats.JSOpt) (*nats.ConsumerInfo, error)
}

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

type bridgeMessage struct {
	subject string
	data    []byte
	ack     func() error
}

type routedBridgeMessage struct {
	sourceIndex  int
	kafka        kafka.Message
	receivedAtMS int64
}

type bridgeSourceState struct {
	source bridgeMessage
	routed bool
	failed bool
}

func runBridge(ctx context.Context, logger *slog.Logger, registry *metrics.Registry, infoReader natsConsumerInfoReader, sub natsPullSubscription, writer kafkaBatchWriter, cfg config) {
	var lastConsumerInfoAt time.Time
	for {
		if ctx.Err() != nil {
			return
		}
		if time.Since(lastConsumerInfoAt) >= time.Duration(envInt("NATS_CONSUMER_METRICS_INTERVAL_SECONDS", 10))*time.Second {
			lastConsumerInfoAt = time.Now()
			if infoReader != nil {
				info, err := infoReader.ConsumerInfo(cfg.NATSStream, cfg.NATSDurable, nats.Context(ctx))
				if err != nil {
					logger.Warn("nats consumer info failed", "stream", cfg.NATSStream, "durable", cfg.NATSDurable, "error", err)
				} else {
					recordNATSConsumerInfoMetrics(registry, cfg, info)
				}
			}
		}
		msgs, err := sub.Fetch(cfg.BatchSize, nats.MaxWait(cfg.FetchWait))
		if err != nil {
			if isBridgeShutdownFetchError(ctx, err) {
				return
			}
			if errors.Is(err, nats.ErrTimeout) {
				continue
			}
			if isTransientBridgeFetchError(err) {
				logger.Warn("nats fetch interrupted", "error", err)
				time.Sleep(time.Second)
				continue
			}
			logger.Error("nats fetch failed", "error", err)
			time.Sleep(time.Second)
			continue
		}
		bridgeMessages := make([]bridgeMessage, 0, len(msgs))
		for _, msg := range msgs {
			msg := msg
			bridgeMessages = append(bridgeMessages, bridgeMessage{
				subject: msg.Subject,
				data:    msg.Data,
				ack: func() error {
					return msg.Ack()
				},
			})
		}
		operationCtx, cancel := context.WithTimeout(context.WithoutCancel(ctx), cfg.OperationWait)
		err = bridgeBatchWithProjectionConcurrency(operationCtx, registry, writer, bridgeMessages, cfg.Route, cfg.RawFieldRoutes, cfg.DeriveFieldsFromRaw, cfg.KafkaWriteConcurrency)
		cancel()
		if err != nil {
			logger.Error("bridge batch failed", "count", len(bridgeMessages), "error", err)
			continue
		}
	}
}

func isBridgeShutdownFetchError(ctx context.Context, err error) bool {
	if err == nil {
		return false
	}
	if ctx.Err() != nil {
		return true
	}
	return errors.Is(err, nats.ErrConnectionClosed)
}

func isTransientBridgeFetchError(err error) bool {
	if err == nil {
		return false
	}
	text := strings.ToLower(strings.TrimSpace(err.Error()))
	return strings.Contains(text, "disconnected during fetch") ||
		strings.Contains(text, "connection closed") ||
		strings.Contains(text, "connection reset") ||
		strings.Contains(text, "broken pipe") ||
		strings.Contains(text, "temporary") ||
		strings.Contains(text, "temporarily") ||
		strings.Contains(text, "timeout")
}

func ensureStream(js nats.JetStreamContext, cfg config) error {
	ctx, cancel := context.WithTimeout(context.Background(), cfg.StreamEnsureWait)
	defer cancel()
	opts := []nats.JSOpt{nats.Context(ctx)}
	stream := &nats.StreamConfig{
		Name:       cfg.NATSStream,
		Subjects:   cfg.NATSSubjects,
		Storage:    nats.FileStorage,
		Retention:  nats.LimitsPolicy,
		MaxAge:     cfg.StreamMaxAge,
		MaxBytes:   cfg.StreamMaxBytes,
		Duplicates: 2 * time.Minute,
	}
	if _, err := js.StreamInfo(cfg.NATSStream, opts...); err == nil {
		_, err = js.UpdateStream(stream, opts...)
		return err
	}
	_, err := js.AddStream(stream, opts...)
	if err == nil {
		return nil
	}
	_, updateErr := js.UpdateStream(stream, opts...)
	if updateErr == nil {
		return nil
	}
	return err
}

func bridgeBatch(ctx context.Context, registry *metrics.Registry, writer kafkaBatchWriter, messages []bridgeMessage, route map[string]string) error {
	return bridgeBatchWithProjectionConcurrency(ctx, registry, writer, messages, route, nil, false, 6)
}

func bridgeBatchWithProjection(
	ctx context.Context,
	registry *metrics.Registry,
	writer kafkaBatchWriter,
	messages []bridgeMessage,
	route map[string]string,
	rawFieldRoutes map[string]fieldsProjectionRoute,
	deriveFieldsFromRaw bool,
) error {
	return bridgeBatchWithProjectionConcurrency(ctx, registry, writer, messages, route, rawFieldRoutes, deriveFieldsFromRaw, 6)
}

func bridgeBatchWithProjectionConcurrency(
	ctx context.Context,
	registry *metrics.Registry,
	writer kafkaBatchWriter,
	messages []bridgeMessage,
	route map[string]string,
	rawFieldRoutes map[string]fieldsProjectionRoute,
	deriveFieldsFromRaw bool,
	kafkaWriteConcurrency int,
) error {
	if len(messages) == 0 {
		return nil
	}
	started := time.Now()
	status := "ok"
	defer func() {
		recordBridgeBatchDuration(registry, status, time.Since(started))
	}()
	routed := make([]routedBridgeMessage, 0, len(messages)*2)
	sources := make([]bridgeSourceState, len(messages))
	for sourceIndex, message := range messages {
		sources[sourceIndex].source = message
		addBridgeSubjectMetric(registry, "vehicle_bridge_messages_total", message.subject, "received")
		topic, ok := route[message.subject]
		if !ok || topic == "" {
			addBridgeSubjectMetric(registry, "vehicle_bridge_messages_total", message.subject, "route_error")
			if message.ack != nil {
				if err := message.ack(); err != nil {
					addBridgeSubjectMetric(registry, "vehicle_bridge_nats_acks_total", message.subject, "error")
					status = "error"
					return fmt.Errorf("ack unrouted nats subject %q: %w", message.subject, err)
				}
				addBridgeSubjectMetric(registry, "vehicle_bridge_nats_acks_total", message.subject, "dropped_route_error")
			}
			continue
		}
		kafkaMessage, receivedAtMS, decoded, decodeErr := kafkaMessageWithEnvelope(topic, message.data)
		sources[sourceIndex].routed = true
		routed = append(routed, routedBridgeMessage{
			sourceIndex:  sourceIndex,
			kafka:        kafkaMessage,
			receivedAtMS: receivedAtMS,
		})
		if !deriveFieldsFromRaw {
			continue
		}
		projection, ok := rawFieldRoutes[message.subject]
		if !ok {
			continue
		}
		fieldsMessage, fieldsReceivedAtMS, fieldCount, projectionStatus, projected := projectRawFields(message.subject, decoded, decodeErr, projection)
		recordBridgeFieldsProjection(registry, projection.Protocol, projectionStatus, fieldCount)
		if !projected {
			continue
		}
		routed = append(routed, routedBridgeMessage{
			sourceIndex:  sourceIndex,
			kafka:        fieldsMessage,
			receivedAtMS: fieldsReceivedAtMS,
		})
	}
	if len(routed) == 0 {
		addBridgeBatchPending(registry, 0, 0)
		return nil
	}
	addBridgeBatchPending(registry, len(messages), len(routed))
	defer addBridgeBatchPending(registry, -len(messages), -len(routed))
	groups := groupRoutedBridgeMessagesByTopic(routed)
	results := writeBridgeKafkaGroups(ctx, writer, groups, kafkaWriteConcurrency)
	var firstErr error
	for _, result := range results {
		group := result.group
		err := result.err
		recordBridgeKafkaWriteDuration(registry, group[0].kafka.Topic, statusFromError(err), result.elapsed)
		if err != nil {
			for _, item := range group {
				addBridgeTopicMetric(registry, "vehicle_bridge_kafka_writes_total", item.kafka.Topic, "error")
				sources[item.sourceIndex].failed = true
			}
			status = "error"
			if firstErr == nil {
				firstErr = fmt.Errorf("kafka write topic %s: %w", group[0].kafka.Topic, err)
			}
			continue
		}
		for _, item := range group {
			addBridgeTopicMetric(registry, "vehicle_bridge_kafka_writes_total", item.kafka.Topic, "ok")
			recordBridgeKafkaE2EDuration(registry, item.kafka.Topic, item.receivedAtMS)
		}
	}
	for i := range sources {
		source := &sources[i]
		if !source.routed || source.failed || source.source.ack == nil {
			continue
		}
		if err := source.source.ack(); err != nil {
			addBridgeSubjectMetric(registry, "vehicle_bridge_nats_acks_total", source.source.subject, "error")
			status = "error"
			return err
		}
		addBridgeSubjectMetric(registry, "vehicle_bridge_nats_acks_total", source.source.subject, "ok")
	}
	return firstErr
}

type bridgeKafkaWriteResult struct {
	group   []routedBridgeMessage
	elapsed time.Duration
	err     error
}

func writeBridgeKafkaGroups(ctx context.Context, writer kafkaBatchWriter, groups [][]routedBridgeMessage, concurrency int) []bridgeKafkaWriteResult {
	if len(groups) == 0 {
		return nil
	}
	if concurrency < 1 {
		concurrency = 1
	}
	if concurrency > len(groups) {
		concurrency = len(groups)
	}
	semaphore := make(chan struct{}, concurrency)
	results := make(chan bridgeKafkaWriteResult, len(groups))
	for _, group := range groups {
		group := group
		semaphore <- struct{}{}
		go func() {
			defer func() { <-semaphore }()
			messages := make([]kafka.Message, 0, len(group))
			for _, item := range group {
				messages = append(messages, item.kafka)
			}
			started := time.Now()
			err := writer.WriteMessages(ctx, messages...)
			results <- bridgeKafkaWriteResult{group: group, elapsed: time.Since(started), err: err}
		}()
	}
	out := make([]bridgeKafkaWriteResult, 0, len(groups))
	for range groups {
		out = append(out, <-results)
	}
	return out
}

func projectRawFields(subject string, raw envelope.FrameEnvelope, decodeErr error, projection fieldsProjectionRoute) (kafka.Message, int64, int, string, bool) {
	if decodeErr != nil {
		return kafka.Message{}, 0, 0, "invalid_json", false
	}
	if _, err := topics.ValidateRawEnvelope(subject, raw); err != nil {
		return kafka.Message{}, 0, 0, "invalid_envelope", false
	}
	if raw.Protocol != projection.Protocol {
		return kafka.Message{}, 0, 0, "protocol_mismatch", false
	}
	if !envelope.IsRealtimeTelemetryFrame(raw) {
		return kafka.Message{}, 0, 0, "skipped_non_realtime", false
	}
	if len(raw.ParsedFields) == 0 {
		return kafka.Message{}, 0, 0, "skipped_missing_fields", false
	}
	fields, ok := realtime.BuildFieldsEnvelope(raw)
	if !ok || len(fields.Fields) == 0 {
		return kafka.Message{}, 0, 0, "skipped_missing_fields", false
	}
	if _, err := topics.ValidateFieldsEnvelope(projection.Topic, fields); err != nil {
		return kafka.Message{}, 0, 0, "invalid_fields_envelope", false
	}
	payload, err := fields.MarshalJSONBytes()
	if err != nil {
		return kafka.Message{}, 0, 0, "marshal_error", false
	}
	result := kafka.Message{Topic: projection.Topic, Key: fields.KafkaKey(), Value: payload}
	return result, fields.ReceivedAtMS, len(fields.Fields), "published", true
}

func groupRoutedBridgeMessagesByTopic(messages []routedBridgeMessage) [][]routedBridgeMessage {
	if len(messages) == 0 {
		return nil
	}
	groupsByTopic := make(map[string][]routedBridgeMessage)
	for _, message := range messages {
		groupsByTopic[message.kafka.Topic] = append(groupsByTopic[message.kafka.Topic], message)
	}
	topics := make([]string, 0, len(groupsByTopic))
	for topic := range groupsByTopic {
		topics = append(topics, topic)
	}
	sort.Strings(topics)
	groups := make([][]routedBridgeMessage, 0, len(topics))
	for _, topic := range topics {
		groups = append(groups, groupsByTopic[topic])
	}
	return groups
}

var bridgeBatchDurationBucketsMS = []float64{1, 5, 10, 25, 50, 100, 250, 500, 1000, 5000}
var bridgeKafkaWriteDurationBucketsMS = []float64{1, 5, 10, 25, 50, 100, 250, 500, 1000, 5000}
var bridgeKafkaE2EDurationBucketsMS = []float64{10, 25, 50, 100, 250, 500, 1000, 2500, 5000, 10000}
var bridgeKafkaE2ERecent = metrics.NewRecentLatencyByKey(512)
var bridgeBatchPending = metrics.PendingPairGauge{}

func addBridgeSubjectMetric(registry *metrics.Registry, name string, subject string, status string) {
	if registry == nil {
		return
	}
	labels := metrics.Labels{"subject": subject, "status": status}
	registry.IncCounter(name, labels)
	if name == "vehicle_bridge_messages_total" {
		metrics.RecordLastActivity(registry, "vehicle_bridge_last_message_unix_seconds", labels)
		return
	}
	if name == "vehicle_bridge_nats_acks_total" {
		metrics.RecordLastActivity(registry, "vehicle_bridge_last_ack_unix_seconds", labels)
	}
}

func addBridgeTopicMetric(registry *metrics.Registry, name string, topic string, status string) {
	if registry == nil {
		return
	}
	labels := metrics.Labels{"topic": topic, "status": status}
	registry.IncCounter(name, labels)
	if name == "vehicle_bridge_kafka_writes_total" {
		metrics.RecordLastActivity(registry, "vehicle_bridge_last_kafka_write_unix_seconds", labels)
	}
}

func recordBridgeFieldsProjection(registry *metrics.Registry, protocol envelope.Protocol, status string, fieldCount int) {
	if registry == nil {
		return
	}
	protocolLabel := strings.TrimSpace(string(protocol))
	if protocolLabel == "" {
		protocolLabel = "unknown"
	}
	if strings.TrimSpace(status) == "" {
		status = "unknown"
	}
	labels := metrics.Labels{"protocol": protocolLabel, "status": status}
	registry.IncCounter("vehicle_bridge_fields_projection_total", labels)
	metrics.RecordLastActivity(registry, "vehicle_bridge_last_fields_projection_unix_seconds", labels)
	if status == "published" {
		registry.SetGauge("vehicle_bridge_fields_projection_count", labels, float64(fieldCount))
	}
}

func addBridgeBatchPending(registry *metrics.Registry, messages int, kafkaMessages int) {
	if registry == nil {
		return
	}
	bridgeBatchPending.Add(registry, "vehicle_bridge_batch_pending_messages", "vehicle_bridge_batch_pending_kafka_messages", messages, kafkaMessages)
}

func recordBridgeBatchDuration(registry *metrics.Registry, status string, elapsed time.Duration) {
	if registry == nil {
		return
	}
	registry.ObserveHistogram("vehicle_bridge_batch_duration_ms_histogram", metrics.Labels{
		"status": status,
	}, bridgeBatchDurationBucketsMS, float64(elapsed.Milliseconds()))
}

func recordBridgeKafkaWriteDuration(registry *metrics.Registry, topic string, status string, elapsed time.Duration) {
	if registry == nil {
		return
	}
	registry.ObserveHistogram("vehicle_bridge_kafka_write_duration_ms_histogram", metrics.Labels{
		"topic":  topic,
		"status": status,
	}, bridgeKafkaWriteDurationBucketsMS, float64(elapsed.Milliseconds()))
}

func recordBridgeKafkaE2EDuration(registry *metrics.Registry, topic string, receivedAtMS int64) {
	if registry == nil || receivedAtMS <= 0 {
		return
	}
	elapsedMS := time.Since(time.UnixMilli(receivedAtMS)).Milliseconds()
	if elapsedMS < 0 {
		elapsedMS = 0
	}
	labels := metrics.Labels{
		"topic": topic,
	}
	registry.ObserveHistogram("vehicle_bridge_kafka_e2e_duration_ms_histogram", labels, bridgeKafkaE2EDurationBucketsMS, float64(elapsedMS))
	p99, samples := bridgeKafkaE2ERecent.Observe(topic, float64(elapsedMS))
	registry.SetGauge("vehicle_bridge_kafka_e2e_recent_p99_ms", labels, p99)
	registry.SetGauge("vehicle_bridge_kafka_e2e_recent_samples", labels, float64(samples))
	metrics.RecordLastActivity(registry, "vehicle_bridge_last_kafka_e2e_unix_seconds", labels)
}

func statusFromError(err error) string {
	if err != nil {
		return "error"
	}
	return "ok"
}

func recordNATSConsumerInfoMetrics(registry *metrics.Registry, cfg config, info *nats.ConsumerInfo) {
	if registry == nil || info == nil {
		return
	}
	labels := metrics.Labels{"stream": cfg.NATSStream, "consumer": cfg.NATSDurable}
	registry.SetGauge("vehicle_bridge_nats_consumer_pending", labels, float64(info.NumPending))
	registry.SetGauge("vehicle_bridge_nats_consumer_ack_pending", labels, float64(info.NumAckPending))
	registry.SetGauge("vehicle_bridge_nats_consumer_waiting", labels, float64(info.NumWaiting))
}

func kafkaMessage(topic string, data []byte) kafka.Message {
	message, _ := kafkaMessageWithReceivedAt(topic, data)
	return message
}

func kafkaMessageWithReceivedAt(topic string, data []byte) (kafka.Message, int64) {
	message, receivedAtMS, _, _ := kafkaMessageWithEnvelope(topic, data)
	return message, receivedAtMS
}

func kafkaMessageWithEnvelope(topic string, data []byte) (kafka.Message, int64, envelope.FrameEnvelope, error) {
	var env envelope.FrameEnvelope
	message := kafka.Message{Topic: topic, Value: data}
	if err := json.Unmarshal(data, &env); err != nil {
		return message, 0, envelope.FrameEnvelope{}, err
	}
	message.Key = env.KafkaKey()
	return message, env.ReceivedAtMS, env, nil
}

func env(key string, fallback string) string {
	value := strings.TrimSpace(os.Getenv(key))
	if value == "" {
		return fallback
	}
	return value
}

func envBool(key string, fallback bool) bool {
	value := strings.TrimSpace(os.Getenv(key))
	if value == "" {
		return fallback
	}
	parsed, err := strconv.ParseBool(value)
	if err != nil {
		return fallback
	}
	return parsed
}

func boolMetric(value bool) float64 {
	if value {
		return 1
	}
	return 0
}

func envOptional(key string) (string, bool) {
	value, ok := os.LookupEnv(key)
	if !ok {
		return "", false
	}
	return strings.TrimSpace(value), true
}

func envInt(key string, fallback int) int {
	value := strings.TrimSpace(os.Getenv(key))
	if value == "" {
		return fallback
	}
	parsed, err := strconv.Atoi(value)
	if err != nil || parsed <= 0 {
		return fallback
	}
	return parsed
}

func envInt64(key string, fallback int64) int64 {
	value := strings.TrimSpace(os.Getenv(key))
	if value == "" {
		return fallback
	}
	parsed, err := strconv.ParseInt(value, 10, 64)
	if err != nil || parsed <= 0 {
		return fallback
	}
	return parsed
}

func splitCSV(value string) []string {
	var out []string
	for _, item := range strings.Split(value, ",") {
		item = strings.TrimSpace(item)
		if item != "" {
			out = append(out, item)
		}
	}
	return out
}

func mapKeys(values map[string]string) []string {
	out := make([]string, 0, len(values))
	for key := range values {
		out = append(out, key)
	}
	return out
}