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

package gateway

import (
	"context"
	"crypto/tls"
	"crypto/x509"
	"encoding/hex"
	"errors"
	"fmt"
	"log/slog"
	"os"
	"strings"
	"time"

	mqtt "github.com/eclipse/paho.mqtt.golang"

	"lingniu-vehicle-ingest/go/vehicle-gateway/internal/envelope"
	"lingniu-vehicle-ingest/go/vehicle-gateway/internal/eventbus"
	"lingniu-vehicle-ingest/go/vehicle-gateway/internal/identity"
	"lingniu-vehicle-ingest/go/vehicle-gateway/internal/protocol/yutongmqtt"
)

type MQTTClientConfig struct {
	EndpointName            string
	Broker                  string
	ClientID                string
	Username                string
	Password                string
	Topics                  []string
	QoS                     byte
	CleanSession            bool
	KeepAlive               time.Duration
	ConnectTimeout          time.Duration
	TLSCACertPath           string
	TLSClientCertPath       string
	TLSClientKeyPath        string
	TLSHostnameVerification bool
	Sink                    eventbus.Sink
	Resolver                identity.Resolver
	Logger                  *slog.Logger
}

type MQTTClient struct {
	cfg    MQTTClientConfig
	client mqtt.Client
}

func NewMQTTClient(cfg MQTTClientConfig) (*MQTTClient, error) {
	if strings.TrimSpace(cfg.Broker) == "" {
		return nil, errors.New("mqtt broker is required")
	}
	if strings.TrimSpace(cfg.ClientID) == "" {
		return nil, errors.New("mqtt client id is required")
	}
	if len(cfg.Topics) == 0 {
		return nil, errors.New("mqtt topics are required")
	}
	if cfg.Sink == nil {
		return nil, errors.New("sink is required")
	}
	if cfg.Logger == nil {
		cfg.Logger = slog.Default()
	}
	if cfg.Resolver == nil {
		cfg.Resolver = identity.NoopResolver{}
	}
	if cfg.EndpointName == "" {
		cfg.EndpointName = "yutong"
	}
	return &MQTTClient{cfg: cfg}, nil
}

func (c *MQTTClient) Start(ctx context.Context) error {
	opts, err := c.buildOptions(ctx)
	if err != nil {
		return err
	}

	c.client = mqtt.NewClient(opts)
	token := c.client.Connect()
	if token.Wait() && token.Error() != nil {
		return token.Error()
	}
	go func() {
		<-ctx.Done()
		if c.client != nil && c.client.IsConnected() {
			c.client.Disconnect(250)
		}
	}()
	return nil
}

func (c *MQTTClient) buildOptions(ctx context.Context) (*mqtt.ClientOptions, error) {
	keepAlive := c.cfg.KeepAlive
	if keepAlive <= 0 {
		keepAlive = 20 * time.Second
	}
	connectTimeout := c.cfg.ConnectTimeout
	if connectTimeout <= 0 {
		connectTimeout = 10 * time.Second
	}
	opts := mqtt.NewClientOptions().
		AddBroker(c.cfg.Broker).
		SetClientID(c.cfg.ClientID).
		SetUsername(c.cfg.Username).
		SetPassword(c.cfg.Password).
		SetCleanSession(c.cfg.CleanSession).
		SetAutoReconnect(true).
		SetConnectRetry(true).
		SetConnectRetryInterval(5 * time.Second).
		SetKeepAlive(keepAlive).
		SetConnectTimeout(connectTimeout)

	opts.SetDefaultPublishHandler(func(_ mqtt.Client, message mqtt.Message) {
		c.handleMessage(ctx, message.Topic(), message.Payload())
	})
	opts.OnConnect = func(client mqtt.Client) {
		for _, topic := range c.cfg.Topics {
			token := client.Subscribe(topic, c.cfg.QoS, nil)
			if token.Wait() && token.Error() != nil {
				c.cfg.Logger.Error("mqtt subscribe failed", "broker", c.cfg.Broker, "topic", topic, "error", token.Error())
				continue
			}
			c.cfg.Logger.Info("mqtt subscribed", "broker", c.cfg.Broker, "topic", topic, "qos", c.cfg.QoS)
		}
	}
	opts.OnConnectionLost = func(_ mqtt.Client, err error) {
		c.cfg.Logger.Warn("mqtt connection lost", "broker", c.cfg.Broker, "error", err)
	}

	tlsConfig, err := c.buildTLSConfig()
	if err != nil {
		return nil, err
	}
	if tlsConfig != nil {
		opts.SetTLSConfig(tlsConfig)
	}
	return opts, nil
}

func (c *MQTTClient) buildTLSConfig() (*tls.Config, error) {
	caPath := strings.TrimSpace(c.cfg.TLSCACertPath)
	certPath := strings.TrimSpace(c.cfg.TLSClientCertPath)
	keyPath := strings.TrimSpace(c.cfg.TLSClientKeyPath)
	if caPath == "" && certPath == "" && keyPath == "" {
		return nil, nil
	}
	config := &tls.Config{
		MinVersion:         tls.VersionTLS12,
		InsecureSkipVerify: !c.cfg.TLSHostnameVerification,
	}
	if caPath != "" {
		caPEM, err := os.ReadFile(caPath)
		if err != nil {
			return nil, fmt.Errorf("read mqtt ca certificate: %w", err)
		}
		roots := x509.NewCertPool()
		if !roots.AppendCertsFromPEM(caPEM) {
			return nil, fmt.Errorf("parse mqtt ca certificate %s", caPath)
		}
		config.RootCAs = roots
	}
	if certPath != "" || keyPath != "" {
		if certPath == "" || keyPath == "" {
			return nil, errors.New("mqtt client certificate and key must be configured together")
		}
		cert, err := tls.LoadX509KeyPair(certPath, keyPath)
		if err != nil {
			return nil, fmt.Errorf("load mqtt client certificate: %w", err)
		}
		config.Certificates = []tls.Certificate{cert}
	}
	return config, nil
}

func (c *MQTTClient) handleMessage(ctx context.Context, topic string, payload []byte) {
	receivedAtMS := time.Now().UnixMilli()
	env, err := yutongmqtt.ParseMessage(c.cfg.EndpointName, topic, payload, receivedAtMS)
	if err != nil {
		env = envelope.FrameEnvelope{
			Protocol:       envelope.ProtocolYutongMQTT,
			MessageID:      "MQTT",
			VehicleKeyHint: "unknown",
			SourceEndpoint: "mqtt://" + c.cfg.EndpointName + topic,
			EventTimeMS:    receivedAtMS,
			ReceivedAtMS:   receivedAtMS,
			RawText:        string(payload),
			RawHex:         strings.ToUpper(hex.EncodeToString(payload)),
			ParseStatus:    envelope.ParseBadFrame,
			ParseError:     err.Error(),
		}
		env.EventID = env.StableEventID()
	} else {
		resolved, resolveErr := c.cfg.Resolver.Resolve(ctx, env)
		if resolveErr != nil {
			c.cfg.Logger.Warn("mqtt identity resolve failed", "topic", topic, "event_id", env.StableEventID(), "error", resolveErr)
			if env.Parsed == nil {
				env.Parsed = map[string]any{}
			}
			env.Parsed["identity"] = map[string]any{"resolved": false, "error": resolveErr.Error()}
			env.ParseStatus = envelope.ParsePartial
		} else {
			env = resolved
		}
	}
	if err := c.cfg.Sink.PublishRaw(ctx, env); err != nil {
		c.cfg.Logger.Error("publish mqtt raw failed", "topic", topic, "event_id", env.StableEventID(), "error", err)
		return
	}
	if env.ParseStatus == envelope.ParseBadFrame {
		return
	}
	if err := c.cfg.Sink.PublishUnified(ctx, env); err != nil {
		c.cfg.Logger.Error("publish mqtt unified failed", "topic", topic, "event_id", env.StableEventID(), "error", err)
	}
}