lingniu-vehicle-ingest/go/vehicle-gateway/internal/topics/topics.go

package topics

import (
	"fmt"
	"sort"
	"strings"

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

const (
	RawGB32960       = "vehicle.raw.go.gb32960.v1"
	RawJT808         = "vehicle.raw.go.jt808.v1"
	RawYutongMQTT    = "vehicle.raw.go.yutong-mqtt.v1"
	FieldsGB32960    = "vehicle.fields.go.gb32960.v1"
	FieldsJT808      = "vehicle.fields.go.jt808.v1"
	FieldsYutongMQTT = "vehicle.fields.go.yutong-mqtt.v1"
	Unified          = "vehicle.event.go.unified.v1"
)

const (
	RawPrefix    = "vehicle.raw."
	FieldsPrefix = "vehicle.fields."
)

func ProtocolForKnownRawTopic(topic string) (string, bool) {
	switch strings.TrimSpace(topic) {
	case RawGB32960:
		return "GB32960", true
	case RawJT808:
		return "JT808", true
	case RawYutongMQTT:
		return "YUTONG_MQTT", true
	default:
		return "", false
	}
}

func ProtocolForKnownFieldsTopic(topic string) (string, bool) {
	switch strings.TrimSpace(topic) {
	case FieldsGB32960:
		return "GB32960", true
	case FieldsJT808:
		return "JT808", true
	case FieldsYutongMQTT:
		return "YUTONG_MQTT", true
	default:
		return "", false
	}
}

func ValidateKnownRawTopicProtocol(topic string, protocol string) error {
	expectedProtocol, ok := ProtocolForKnownRawTopic(topic)
	if !ok {
		return nil
	}
	protocol = strings.TrimSpace(protocol)
	if protocol == expectedProtocol {
		return nil
	}
	if protocol == "" {
		protocol = "UNKNOWN"
	}
	return fmt.Errorf("raw topic %q expects protocol %s, got %s", strings.TrimSpace(topic), expectedProtocol, protocol)
}

func ValidateKnownFieldsTopicProtocol(topic string, protocol string) error {
	expectedProtocol, ok := ProtocolForKnownFieldsTopic(topic)
	if !ok {
		return nil
	}
	protocol = strings.TrimSpace(protocol)
	if protocol == expectedProtocol {
		return nil
	}
	if protocol == "" {
		protocol = "UNKNOWN"
	}
	return fmt.Errorf("fields topic %q expects protocol %s, got %s", strings.TrimSpace(topic), expectedProtocol, protocol)
}

func ValidateRawEnvelope(topic string, env envelope.FrameEnvelope) (string, error) {
	if err := ValidateKnownRawTopicProtocol(topic, string(env.Protocol)); err != nil {
		return "protocol_topic_mismatch", err
	}
	if strings.HasPrefix(strings.TrimSpace(topic), RawPrefix) {
		switch env.EventKind {
		case "", envelope.EventKindRaw:
			return "", nil
		default:
			return "event_kind_mismatch", fmt.Errorf("raw topic %q expects event kind %s, got %s", strings.TrimSpace(topic), envelope.EventKindRaw, env.EventKind)
		}
	}
	return "", nil
}

func ValidateFieldsEnvelope(topic string, env envelope.FrameEnvelope) (string, error) {
	if err := ValidateKnownFieldsTopicProtocol(topic, string(env.Protocol)); err != nil {
		return "protocol_topic_mismatch", err
	}
	if strings.HasPrefix(strings.TrimSpace(topic), FieldsPrefix) {
		if env.EventKind != envelope.EventKindFields {
			return "event_kind_mismatch", fmt.Errorf("fields topic %q expects event kind %s, got %s", strings.TrimSpace(topic), envelope.EventKindFields, env.EventKind)
		}
		if strings.TrimSpace(env.FieldMapping) == "" {
			return "missing_field_mapping", fmt.Errorf("fields topic %q expects non-empty field mapping", strings.TrimSpace(topic))
		}
		if len(env.Fields) == 0 {
			return "missing_fields", fmt.Errorf("fields topic %q expects non-empty fields payload", strings.TrimSpace(topic))
		}
		if err := ValidateProtocolFieldNames(env.Protocol, env.Fields); err != nil {
			return "invalid_field_name", err
		}
	}
	return "", nil
}

func ValidateProtocolFieldNames(protocol envelope.Protocol, fields map[string]any) error {
	prefix, ok := protocolFieldPrefix(protocol)
	if !ok {
		return nil
	}
	invalid := make([]string, 0)
	for key := range fields {
		trimmed := strings.TrimSpace(key)
		if key == trimmed && len(trimmed) > len(prefix) && strings.HasPrefix(trimmed, prefix) {
			continue
		}
		invalid = append(invalid, key)
	}
	if len(invalid) == 0 {
		return nil
	}
	sort.Strings(invalid)
	if len(invalid) > 3 {
		invalid = invalid[:3]
	}
	return fmt.Errorf("fields protocol %s expects every field name under %q; invalid fields: %q", protocol, prefix, invalid)
}

func protocolFieldPrefix(protocol envelope.Protocol) (string, bool) {
	switch protocol {
	case envelope.ProtocolGB32960:
		return "gb32960.", true
	case envelope.ProtocolJT808:
		return "jt808.", true
	case envelope.ProtocolYutongMQTT:
		return "yutong_mqtt.", true
	default:
		return "", false
	}
}

func ValidateKnownRawFieldsProtocols(raw map[string]string, fields map[string]string, valueName string) error {
	for protocol, value := range raw {
		if err := ValidateKnownRawTopicProtocol(value, protocol); err != nil {
			return fmt.Errorf("raw %s for %s must match protocol: %w", valueName, protocol, err)
		}
	}
	for protocol, value := range fields {
		if err := ValidateKnownFieldsTopicProtocol(value, protocol); err != nil {
			return fmt.Errorf("fields %s for %s must match protocol: %w", valueName, protocol, err)
		}
	}
	return nil
}

func ValidateKafkaRawFields(raw map[string]string, fields map[string]string) error {
	for name, topic := range raw {
		topic = strings.TrimSpace(topic)
		if topic == "" {
			return fmt.Errorf("raw kafka topic for %s is empty", name)
		}
		if !strings.HasPrefix(topic, RawPrefix) {
			return fmt.Errorf("raw kafka topic for %s must start with %q, got %q", name, RawPrefix, topic)
		}
	}
	for name, topic := range fields {
		topic = strings.TrimSpace(topic)
		if topic == "" {
			return fmt.Errorf("fields kafka topic for %s is empty", name)
		}
		if !strings.HasPrefix(topic, FieldsPrefix) {
			return fmt.Errorf("fields kafka topic for %s must start with %q, got %q", name, FieldsPrefix, topic)
		}
	}
	if err := ValidateKnownRawFieldsProtocols(raw, fields, "kafka topic"); err != nil {
		return err
	}
	return ValidateRawFieldsDisjoint(raw, fields, "kafka topic")
}

func ValidateRawFieldsDisjoint(raw map[string]string, fields map[string]string, valueName string) error {
	rawByValue := map[string]string{}
	for name, value := range raw {
		value = strings.TrimSpace(value)
		if value == "" {
			continue
		}
		rawByValue[value] = name
	}
	for fieldsName, value := range fields {
		value = strings.TrimSpace(value)
		if value == "" {
			continue
		}
		if rawName, ok := rawByValue[value]; ok {
			return fmt.Errorf("raw and fields %s must be different: raw %s and fields %s both use %q", valueName, rawName, fieldsName, value)
		}
	}
	return nil
}