✨ feat(mqtt): 新增 MQTT 数据源连接与测试发消息支持

internal/app/db_context.go

@@ -16,6 +16,8 @@ func normalizeRunConfig(config connection.ConnectionConfig, dbName string) conne
 	}
 
 	switch strings.ToLower(strings.TrimSpace(config.Type)) {
+	case "mqtt", "mqtts":
+		// MQTT 的 Database 字段表示默认 Topic，不能被树上的 synthetic database(topics) 覆盖。
 	case "kafka", "apache-kafka", "apache_kafka":
 		// Kafka 的 Database 字段表示默认 Topic，不能被树上的 synthetic database(topics) 覆盖。
 	case "oceanbase":
@@ -53,7 +55,7 @@ func normalizeSchemaAndTable(config connection.ConnectionConfig, dbName string,
 
 	// Elasticsearch：索引名可能含多个点（如 iot_pro_biz_operate_log.index.20240626），
 	// 不能按点分割，直接返回原始数据库名和完整表名。
-	if dbType == "elasticsearch" || dbType == "iotdb" || dbType == "kafka" || dbType == "rabbitmq" {
+	if dbType == "elasticsearch" || dbType == "iotdb" || dbType == "mqtt" || dbType == "kafka" || dbType == "rabbitmq" {
 		return rawDB, rawTable
 	}
 
@@ -112,7 +114,7 @@ func normalizeSchemaAndTable(config connection.ConnectionConfig, dbName string,
 func normalizeMetadataSchemaAndTable(config connection.ConnectionConfig, dbName string, tableName string) (string, string) {
 	schema, table := normalizeSchemaAndTable(config, dbName, tableName)
 	switch resolveDDLDBType(config) {
-	case "kafka", "rabbitmq":
+	case "mqtt", "kafka", "rabbitmq":
 		return schema, table
 	case "postgres", "kingbase", "highgo", "vastbase", "opengauss", "gaussdb":
 		rawTable := strings.TrimSpace(tableName)

internal/app/db_context_test.go

@@ -332,6 +332,18 @@ func TestNormalizeSchemaAndTable_KafkaPreservesDottedTopicName(t *testing.T) {
 	}
 }
 
+func TestNormalizeSchemaAndTable_MQTTPreservesTopicFilter(t *testing.T) {
+	t.Parallel()
+
+	schemaOrDb, table := normalizeSchemaAndTable(connection.ConnectionConfig{
+		Type: "mqtt",
+	}, "topics", "devices/floor1.sensor.v1")
+
+	if schemaOrDb != "topics" || table != "devices/floor1.sensor.v1" {
+		t.Fatalf("expected mqtt topic filter to stay intact, got %q.%q", schemaOrDb, table)
+	}
+}
+
 func TestNormalizeSchemaAndTable_RabbitMQPreservesDottedQueueName(t *testing.T) {
 	t.Parallel()
 
@@ -356,6 +368,18 @@ func TestNormalizeMetadataSchemaAndTable_KafkaPreservesDottedTopicName(t *testin
 	}
 }
 
+func TestNormalizeMetadataSchemaAndTable_MQTTPreservesTopicFilter(t *testing.T) {
+	t.Parallel()
+
+	schemaOrDb, table := normalizeMetadataSchemaAndTable(connection.ConnectionConfig{
+		Type: "mqtt",
+	}, "topics", "devices/floor1.sensor.v1")
+
+	if schemaOrDb != "topics" || table != "devices/floor1.sensor.v1" {
+		t.Fatalf("expected mqtt metadata topic filter to stay intact, got %q.%q", schemaOrDb, table)
+	}
+}
+
 func TestNormalizeMetadataSchemaAndTable_RabbitMQPreservesDottedQueueName(t *testing.T) {
 	t.Parallel()
 

internal/app/methods_db.go

@@ -315,8 +315,8 @@ func normalizeSchemaAndTableByType(dbType string, dbName string, tableName strin
 		return rawDB, rawTable
 	}
 
-	// Elasticsearch / RabbitMQ / Kafka：对象名可能含多个点，不能按点分割
-	if dbType == "elasticsearch" || dbType == "kafka" || dbType == "rabbitmq" {
+	// Elasticsearch / MQTT / RabbitMQ / Kafka：对象名可能含多个点或路径，不能按点分割
+	if dbType == "elasticsearch" || dbType == "mqtt" || dbType == "kafka" || dbType == "rabbitmq" {
 		return rawDB, rawTable
 	}
 

internal/app/methods_db_create_statement_test.go

@@ -166,6 +166,15 @@ func TestNormalizeSchemaAndTableByType_KafkaPreservesDottedTopicName(t *testing.
 	}
 }
 
+func TestNormalizeSchemaAndTableByType_MQTTPreservesTopicFilter(t *testing.T) {
+	t.Parallel()
+
+	schema, table := normalizeSchemaAndTableByType("mqtt", "topics", "devices/floor1.sensor.v1")
+	if schema != "topics" || table != "devices/floor1.sensor.v1" {
+		t.Fatalf("expected mqtt topic filter to stay intact, got %q.%q", schema, table)
+	}
+}
+
 func TestNormalizeSchemaAndTableByType_RabbitMQPreservesDottedQueueName(t *testing.T) {
 	t.Parallel()
 

internal/app/methods_driver.go

@@ -1432,6 +1432,8 @@ func normalizeDriverType(driverType string) string {
 		return "gaussdb"
 	case "goldendb", "greatdb", "gdb":
 		return "goldendb"
+	case "mqtt", "mqtts":
+		return "mqtt"
 	case "kafka", "apache-kafka", "apache_kafka":
 		return "kafka"
 	case "rabbitmq", "rabbit-mq", "rabbit_mq":
@@ -1505,6 +1507,7 @@ func allDriverDefinitionsWithPackages(packages map[string]pinnedDriverPackage) [
 		{Type: "oracle", Name: "Oracle", Engine: driverEngineGo, BuiltIn: true},
 		{Type: "redis", Name: "Redis", Engine: driverEngineGo, BuiltIn: true},
 		{Type: "postgres", Name: "PostgreSQL", Engine: driverEngineGo, BuiltIn: true},
+		{Type: "mqtt", Name: "MQTT", Engine: driverEngineGo, BuiltIn: true},
 		{Type: "kafka", Name: "Kafka", Engine: driverEngineGo, BuiltIn: true},
 		{Type: "rabbitmq", Name: "RabbitMQ", Engine: driverEngineGo, BuiltIn: true},
 

internal/app/methods_driver_version_test.go

@@ -518,6 +518,22 @@ func TestKafkaDriverDefinitionIsBuiltIn(t *testing.T) {
 	}
 }
 
+func TestMQTTDriverDefinitionIsBuiltIn(t *testing.T) {
+	definition, ok := resolveDriverDefinition("mqtts")
+	if !ok {
+		t.Fatal("expected mqtt driver definition")
+	}
+	if definition.Name != "MQTT" {
+		t.Fatalf("unexpected mqtt driver name: %q", definition.Name)
+	}
+	if !definition.BuiltIn {
+		t.Fatal("expected mqtt to be a built-in driver")
+	}
+	if definition.PinnedVersion != "" || definition.DefaultDownloadURL != "" {
+		t.Fatalf("expected mqtt builtin definition to omit optional-agent metadata: %#v", definition)
+	}
+}
+
 func TestRabbitMQDriverDefinitionIsBuiltIn(t *testing.T) {
 	definition, ok := resolveDriverDefinition("rabbit-mq")
 	if !ok {

internal/db/database.go

@@ -489,6 +489,9 @@ var databaseFactories = map[string]databaseFactory{
 	"qdrant": func() Database {
 		return &QdrantDB{}
 	},
+	"mqtt": func() Database {
+		return &MQTTDB{}
+	},
 	"kafka": func() Database {
 		return &KafkaDB{}
 	},
@@ -535,6 +538,8 @@ func normalizeDatabaseType(dbType string) string {
 		return "chroma"
 	case "qdrantdb", "qdrant-db":
 		return "qdrant"
+	case "mqtt", "mqtts":
+		return "mqtt"
 	case "kafka", "apache-kafka", "apache_kafka":
 		return "kafka"
 	case "rabbitmq", "rabbit-mq", "rabbit_mq":

internal/db/driver_support.go

@@ -19,6 +19,7 @@ var coreBuiltinDrivers = map[string]struct{}{
 	"postgres": {},
 	"chroma":   {},
 	"qdrant":   {},
+	"mqtt":     {},
 	"kafka":    {},
 	"rabbitmq": {},
 }
@@ -81,6 +82,8 @@ func normalizeRuntimeDriverType(driverType string) string {
 		return "chroma"
 	case "qdrantdb", "qdrant-db":
 		return "qdrant"
+	case "mqtt", "mqtts":
+		return "mqtt"
 	case "apache-iotdb", "apache_iotdb", "iotdb":
 		return "iotdb"
 	case "kafka", "apache-kafka", "apache_kafka":
@@ -148,6 +151,8 @@ func driverDisplayName(driverType string) string {
 		return "Chroma"
 	case "qdrant":
 		return "Qdrant"
+	case "mqtt":
+		return "MQTT"
 	case "kafka":
 		return "Kafka"
 	case "rabbitmq":

internal/db/mqtt_impl_test.go

@@ -0,0 +1,211 @@
+package db
+
+import (
+	"context"
+	"reflect"
+	"strings"
+	"testing"
+	"time"
+
+	"GoNavi-Wails/internal/connection"
+)
+
+func TestNormalizeMQTTConfigParsesURIAndParams(t *testing.T) {
+	config := normalizeMQTTConfig(connection.ConnectionConfig{
+		URI:              "mqtt://user:secret@127.0.0.1:1883/devices%2F%2B%2Ftelemetry?topology=cluster&tls=true&skip_verify=true",
+		ConnectionParams: "topics=devices%2F%2B%2Ftelemetry,%24SYS%2F%23&qos=1&retain=false&cleanSession=false&fetchWaitMs=3500",
+	})
+
+	if config.Host != "127.0.0.1" || config.Port != 1883 {
+		t.Fatalf("unexpected mqtt host/port: %#v", config)
+	}
+	if config.User != "user" || config.Password != "secret" {
+		t.Fatalf("unexpected mqtt credentials: %#v", config)
+	}
+	if config.Database != "devices/+/telemetry" {
+		t.Fatalf("unexpected mqtt default topic: %q", config.Database)
+	}
+	if !config.UseSSL || config.SSLMode != "skip-verify" {
+		t.Fatalf("unexpected mqtt tls settings: %#v", config)
+	}
+	if config.Topology != "cluster" {
+		t.Fatalf("unexpected mqtt topology: %q", config.Topology)
+	}
+
+	params := mqttConnectionParams(config)
+	if params.Get("topics") != "devices/+/telemetry,$SYS/#" {
+		t.Fatalf("unexpected mqtt topics param: %#v", params)
+	}
+	if params.Get("qos") != "1" || params.Get("fetchWaitMs") != "3500" {
+		t.Fatalf("unexpected mqtt params: %#v", params)
+	}
+}
+
+func TestMQTTQueryExecAndColumns(t *testing.T) {
+	fakeRuntime := &fakeMQTTRuntime{
+		fetchResponses: map[string][]mqttMessageRecord{
+			"devices/+/telemetry": {
+				{
+					Topic:      "devices/device-001/telemetry",
+					QoS:        1,
+					Retained:   false,
+					Duplicate:  false,
+					MessageID:  12,
+					Payload:    []byte(`{"event":"created","meta":{"source":"sensor"}}`),
+					Decoded:    map[string]interface{}{"event": "created", "meta": map[string]interface{}{"source": "sensor"}},
+					Encoding:   "json",
+					ReceivedAt: time.Date(2026, 6, 14, 11, 0, 0, 0, time.UTC),
+				},
+				{
+					Topic:      "devices/device-002/telemetry",
+					QoS:        1,
+					Retained:   true,
+					Duplicate:  false,
+					MessageID:  13,
+					Payload:    []byte("plain-text"),
+					Decoded:    "plain-text",
+					Encoding:   "text",
+					ReceivedAt: time.Date(2026, 6, 14, 11, 0, 1, 0, time.UTC),
+				},
+			},
+		},
+	}
+
+	originalFactory := newMQTTRuntime
+	newMQTTRuntime = func(config connection.ConnectionConfig) (mqttRuntime, error) {
+		return fakeRuntime, nil
+	}
+	defer func() {
+		newMQTTRuntime = originalFactory
+	}()
+
+	client := &MQTTDB{}
+	if err := client.Connect(connection.ConnectionConfig{
+		Type:             "mqtt",
+		Host:             "127.0.0.1",
+		Port:             1883,
+		Database:         "devices/+/telemetry",
+		ConnectionParams: "topics=devices%2F%2B%2Ftelemetry,%24SYS%2F%23&qos=1&fetchWaitMs=2500",
+	}); err != nil {
+		t.Fatalf("Connect failed: %v", err)
+	}
+	defer client.Close()
+
+	rows, columns, err := client.Query(`SHOW TOPICS LIMIT 2`)
+	if err != nil {
+		t.Fatalf("SHOW TOPICS failed: %v", err)
+	}
+	if len(rows) != 2 || rows[0]["topic"] != "devices/+/telemetry" {
+		t.Fatalf("unexpected mqtt topic rows: %#v", rows)
+	}
+	if !containsString(columns, "wildcard") {
+		t.Fatalf("expected wildcard column, got %v", columns)
+	}
+
+	rows, _, err = client.Query(`DESCRIBE TOPIC "devices/+/telemetry"`)
+	if err != nil {
+		t.Fatalf("DESCRIBE TOPIC failed: %v", err)
+	}
+	if len(rows) != 1 || rows[0]["configured"] != true || rows[0]["default_qos"] != 1 {
+		t.Fatalf("unexpected mqtt describe rows: %#v", rows)
+	}
+
+	rows, columns, err = client.Query(`SELECT * FROM "devices/+/telemetry" LIMIT 1 OFFSET 1`)
+	if err != nil {
+		t.Fatalf("SELECT topic failed: %v", err)
+	}
+	if len(fakeRuntime.fetchRequests) == 0 || fakeRuntime.fetchRequests[len(fakeRuntime.fetchRequests)-1].Offset != 1 {
+		t.Fatalf("expected mqtt fetch offset 1, got %#v", fakeRuntime.fetchRequests)
+	}
+	if len(rows) != 1 || rows[0]["payload"] != "plain-text" || rows[0]["payload_encoding"] != "text" {
+		t.Fatalf("unexpected mqtt message rows: %#v", rows)
+	}
+	if !containsString(columns, "payload_encoding") {
+		t.Fatalf("expected payload_encoding column, got %v", columns)
+	}
+
+	affected, err := client.Exec(`{"publish":"devices/device-001/telemetry","payload":{"id":1},"qos":2,"retain":true}`)
+	if err != nil {
+		t.Fatalf("mqtt publish failed: %v", err)
+	}
+	if affected != 1 {
+		t.Fatalf("unexpected affected rows: %d", affected)
+	}
+	if len(fakeRuntime.published) != 1 {
+		t.Fatalf("expected one mqtt publish call, got %#v", fakeRuntime.published)
+	}
+	if fakeRuntime.published[0].Topic != "devices/device-001/telemetry" || fakeRuntime.published[0].QoS != 2 || !fakeRuntime.published[0].Retain {
+		t.Fatalf("unexpected mqtt publish command: %#v", fakeRuntime.published[0])
+	}
+
+	columnDefs, err := client.GetColumns(mqttSyntheticDatabase, "devices/+/telemetry")
+	if err != nil {
+		t.Fatalf("GetColumns failed: %v", err)
+	}
+	names := make([]string, 0, len(columnDefs))
+	for _, col := range columnDefs {
+		names = append(names, col.Name)
+	}
+	joined := strings.Join(names, ",")
+	for _, want := range []string{"topic", "payload.meta.source", "payload_encoding"} {
+		if !strings.Contains(joined, want) {
+			t.Fatalf("expected mqtt column %q in %s", want, joined)
+		}
+	}
+
+	databases, err := client.GetDatabases()
+	if err != nil {
+		t.Fatalf("GetDatabases failed: %v", err)
+	}
+	if !reflect.DeepEqual(databases, []string{mqttSyntheticDatabase}) {
+		t.Fatalf("unexpected mqtt database list: %#v", databases)
+	}
+
+	tables, err := client.GetTables(mqttSyntheticDatabase)
+	if err != nil {
+		t.Fatalf("GetTables failed: %v", err)
+	}
+	if !reflect.DeepEqual(tables, []string{"$SYS/#", "devices/+/telemetry"}) {
+		t.Fatalf("unexpected mqtt topic list: %#v", tables)
+	}
+
+	if _, _, err := client.Query(`SELECT COUNT(*) FROM "devices/+/telemetry"`); err == nil || !strings.Contains(err.Error(), "COUNT(*)") {
+		t.Fatalf("expected COUNT(*) to be rejected, got %v", err)
+	}
+}
+
+type fakeMQTTRuntime struct {
+	fetchResponses map[string][]mqttMessageRecord
+	fetchRequests  []mqttFetchRequest
+	published      []mqttPublishCommand
+	closed         bool
+}
+
+func (f *fakeMQTTRuntime) Close() error {
+	f.closed = true
+	return nil
+}
+
+func (f *fakeMQTTRuntime) Ping(ctx context.Context) error {
+	return nil
+}
+
+func (f *fakeMQTTRuntime) FetchMessages(ctx context.Context, request mqttFetchRequest) ([]mqttMessageRecord, error) {
+	f.fetchRequests = append(f.fetchRequests, request)
+	items := append([]mqttMessageRecord(nil), f.fetchResponses[request.Topic]...)
+	if request.Offset > 0 {
+		if request.Offset >= len(items) {
+			return []mqttMessageRecord{}, nil
+		}
+		items = items[request.Offset:]
+	}
+	if request.Limit > 0 && len(items) > request.Limit {
+		items = items[:request.Limit]
+	}
+	return items, nil
+}
+
+func (f *fakeMQTTRuntime) Publish(ctx context.Context, command mqttPublishCommand) (int64, error) {
+	f.published = append(f.published, command)
+	return 1, nil
+}