From 7c6d854e0119d18516d806da79dc52c2f124fc99 Mon Sep 17 00:00:00 2001 From: Live Control Date: Sun, 15 Mar 2026 15:34:52 +0300 Subject: [PATCH] channel mqtt --- config/config.example.json | 24 +++ docs/channels/mqtt/README.md | 80 +++++++++ go.mod | 8 + go.sum | 2 + pkg/channels/manager.go | 4 + pkg/channels/mqtt/init.go | 16 ++ pkg/channels/mqtt/mqtt.go | 314 +++++++++++++++++++++++++++++++++++ pkg/config/config.go | 22 +++ pkg/gateway/gateway.go | 1 + 9 files changed, 471 insertions(+) create mode 100644 docs/channels/mqtt/README.md create mode 100644 pkg/channels/mqtt/init.go create mode 100644 pkg/channels/mqtt/mqtt.go diff --git a/config/config.example.json b/config/config.example.json index 88578701a..51b19110d 100644 --- a/config/config.example.json +++ b/config/config.example.json @@ -262,6 +262,30 @@ "enabled": false }, "reasoning_channel_id": "" + }, + "mqtt": { + "enabled": false, + "broker": "tcp://broker.emqx.io:1883", + "client_id": "picoclaw-001", + "username": "", + "password": "", + "subscribe_topics": [ + "picoclaw/input" + ], + "reply_topic": "picoclaw/output", + "tls": false, + "tls_ca": "", + "tls_cert": "", + "tls_key": "", + "qos": 1, + "retain": false, + "prefix": "/mqtt", + "instruction": "Будь предельно краток", + "allow_from": [], + "group_trigger": { + "mention_only": false + }, + "reasoning_channel_id": "" } }, "providers": { diff --git a/docs/channels/mqtt/README.md b/docs/channels/mqtt/README.md new file mode 100644 index 000000000..370c44ffa --- /dev/null +++ b/docs/channels/mqtt/README.md @@ -0,0 +1,80 @@ +# MQTT Channel Configuration Guide + +## 1. Example Configuration + +Add this to `config.json`: + +```json +{ + "channels": { + "mqtt": { + "enabled": true, + "broker": "tcp://localhost:1883", + "client_id": "picoclaw-bot", + "username": "", + "password": "", + "qos": 1, + "retain": false, + "tls": false, + "subscribe_topics": ["picoclaw/chat"], + "reply_topic": "picoclaw/reply", + "allow_from": [], + "group_trigger": { + "mention_only": true + }, + "reasoning_channel_id": "", + "instruction": "" + } + } +} +``` + +## 2. Field Reference + +| Field | Type | Required | Description | +|----------------------|----------|----------|-------------| +| enabled | bool | Yes | Enable or disable the MQTT channel | +| broker | string | Yes | MQTT broker URL (e.g., `tcp://localhost:1883`, `ssl://broker.example.com:8883`) | +| client_id | string | Yes | Unique client identifier for the MQTT connection | +| username | string | No | MQTT username for authentication | +| password | string | No | MQTT password for authentication | +| qos | int | No | Quality of Service level (0, 1, or 2). Default: 1 | +| retain | bool | No | Whether to retain messages. Default: false | +| tls | bool | No | Enable TLS/SSL connection. Default: false | +| subscribe_topics | []string | Yes | List of MQTT topics to subscribe to for incoming messages | +| reply_topic | string | No | Topic to publish replies to. Supports placeholders: `{client_id}`, `{topic}` | +| allow_from | []string | No | Client ID whitelist (empty allows all) | +| group_trigger | object | No | Group trigger strategy (`mention_only` / `prefixes`) | +| reasoning_channel_id | string | No | Target channel for reasoning output | +| instruction | string | No | Optional instruction prefix added to all incoming messages | + +## 3. Currently Supported + +- **Message Format**: Supports both JSON and plain text messages + - JSON format: `{"status": "your message"}` + - Plain text: Direct text content + - Automatic JSON parsing with fallback to plain text for malformed JSON +- **Authentication**: Username/password authentication support +- **TLS/SSL**: Secure connections with TLS configuration +- **Quality of Service**: Configurable QoS levels (0, 1, 2) +- **Topic Management**: Multiple subscribe topics and configurable reply topics +- **Message Retention**: Optional message retention on broker +- **Auto-reconnection**: Automatic reconnection with exponential backoff +- **Group Triggers**: Support for mention-only and prefix-based triggers +- **Placeholder Replacement**: Dynamic topic names using `{client_id}` and `{topic}` placeholders + +## 4. Features + +- **Robust Message Handling**: Intelligent parsing that handles malformed JSON gracefully +- **Flexible Topic Configuration**: Support for multiple input topics and dynamic reply topics +- **Connection Resilience**: Automatic reconnection with configurable retry intervals +- **Security**: TLS support and authentication for secure communication +- **Message Routing**: Support for reasoning channel routing and group trigger rules + +## 5. Usage Notes + +- The channel automatically handles malformed JSON by attempting to clean and parse it +- Reply topics can use placeholders to dynamically route responses +- Client IDs are used as sender identifiers in the messaging system +- Topics are treated as channels for message routing purposes +- The instruction field allows adding context or commands to all incoming messages \ No newline at end of file diff --git a/go.mod b/go.mod index cfc930d37..1376ae687 100644 --- a/go.mod +++ b/go.mod @@ -10,6 +10,7 @@ require ( github.com/bwmarrin/discordgo v0.29.0 github.com/caarlos0/env/v11 v11.4.0 github.com/ergochat/irc-go v0.6.0 + github.com/eclipse/paho.mqtt.golang v1.4.3 github.com/ergochat/readline v0.1.3 github.com/gdamore/tcell/v2 v2.13.8 github.com/gomarkdown/markdown v0.0.0-20260217112301-37c66b85d6ab @@ -93,8 +94,15 @@ require ( github.com/valyala/fastjson v1.6.10 // indirect github.com/yosida95/uritemplate/v3 v3.0.2 // indirect golang.org/x/arch v0.24.0 // indirect +<<<<<<< HEAD golang.org/x/crypto v0.49.0 golang.org/x/net v0.52.0 golang.org/x/sync v0.20.0 // indirect golang.org/x/sys v0.42.0 // indirect +======= + golang.org/x/crypto v0.48.0 + golang.org/x/net v0.51.0 + golang.org/x/sync v0.19.0 // indirect + golang.org/x/sys v0.41.0 // indirect +>>>>>>> 1038a05 (channel mqtt) ) diff --git a/go.sum b/go.sum index f24b997d4..cb338e77e 100644 --- a/go.sum +++ b/go.sum @@ -46,6 +46,8 @@ github.com/dnaeon/go-vcr v1.2.0 h1:zHCHvJYTMh1N7xnV7zf1m1GPBF9Ad0Jk/whtQ1663qI= github.com/dnaeon/go-vcr v1.2.0/go.mod h1:R4UdLID7HZT3taECzJs4YgbbH6PIGXB6W/sc5OLb6RQ= github.com/dustin/go-humanize v1.0.1 h1:GzkhY7T5VNhEkwH0PVJgjz+fX1rhBrR7pRT3mDkpeCY= github.com/dustin/go-humanize v1.0.1/go.mod h1:Mu1zIs6XwVuF/gI1OepvI0qD18qycQx+mFykh5fBlto= +github.com/eclipse/paho.mqtt.golang v1.4.3 h1:2kwcUGn8seMUfWndX0hGbvH8r7crgcJguQNCyp70xik= +github.com/eclipse/paho.mqtt.golang v1.4.3/go.mod h1:CSYvoAlsMkhYOXh/oKyxa8EcBci6dVkLCbo5tTC1RIE= github.com/elliotchance/orderedmap/v3 v3.1.0 h1:j4DJ5ObEmMBt/lcwIecKcoRxIQUEnw0L804lXYDt/pg= github.com/elliotchance/orderedmap/v3 v3.1.0/go.mod h1:G+Hc2RwaZvJMcS4JpGCOyViCnGeKf0bTYCGTO4uhjSo= github.com/ergochat/irc-go v0.6.0 h1:Y0AGV76aeihJfCtLaQh+OyJKFiKGrYC0VTkeMZ6XW28= diff --git a/pkg/channels/manager.go b/pkg/channels/manager.go index dd0b129e4..d88f20aa3 100644 --- a/pkg/channels/manager.go +++ b/pkg/channels/manager.go @@ -401,6 +401,10 @@ func (m *Manager) initChannels(channels *config.ChannelsConfig) error { m.initChannel("irc", "IRC") } + if m.config.Channels.MQTT.Enabled && m.config.Channels.MQTT.Broker != "" && m.config.Channels.MQTT.ClientID != "" && len(m.config.Channels.MQTT.SubscribeTopics) > 0 { + m.initChannel("mqtt", "MQTT") + } + logger.InfoCF("channels", "Channel initialization completed", map[string]any{ "enabled_channels": len(m.channels), }) diff --git a/pkg/channels/mqtt/init.go b/pkg/channels/mqtt/init.go new file mode 100644 index 000000000..c5f77713f --- /dev/null +++ b/pkg/channels/mqtt/init.go @@ -0,0 +1,16 @@ +package mqtt + +import ( + "github.com/sipeed/picoclaw/pkg/bus" + "github.com/sipeed/picoclaw/pkg/channels" + "github.com/sipeed/picoclaw/pkg/config" +) + +func init() { + channels.RegisterFactory("mqtt", func(cfg *config.Config, b *bus.MessageBus) (channels.Channel, error) { + if !cfg.Channels.MQTT.Enabled { + return nil, nil + } + return NewMQTTChannel(cfg.Channels.MQTT, b) + }) +} \ No newline at end of file diff --git a/pkg/channels/mqtt/mqtt.go b/pkg/channels/mqtt/mqtt.go new file mode 100644 index 000000000..d6efe040d --- /dev/null +++ b/pkg/channels/mqtt/mqtt.go @@ -0,0 +1,314 @@ +package mqtt + +import ( + "context" + "crypto/tls" + "encoding/json" + "fmt" + "strings" + "time" + + mqtt "github.com/eclipse/paho.mqtt.golang" + + "github.com/sipeed/picoclaw/pkg/bus" + "github.com/sipeed/picoclaw/pkg/channels" + "github.com/sipeed/picoclaw/pkg/config" + "github.com/sipeed/picoclaw/pkg/logger" +) + +// MQTTChannel implements the Channel interface for MQTT brokers. +type MQTTChannel struct { + *channels.BaseChannel + config config.MQTTConfig + client mqtt.Client + ctx context.Context + cancel context.CancelFunc +} + +// MQTTMessage represents the JSON structure for MQTT messages. +type MQTTMessage struct { + Status string `json:"status"` +} + +// NewMQTTChannel creates a new MQTT channel. +func NewMQTTChannel(cfg config.MQTTConfig, messageBus *bus.MessageBus) (*MQTTChannel, error) { + if cfg.Broker == "" { + return nil, fmt.Errorf("mqtt broker is required") + } + if cfg.ClientID == "" { + return nil, fmt.Errorf("mqtt client_id is required") + } + if len(cfg.SubscribeTopics) == 0 { + return nil, fmt.Errorf("mqtt subscribe_topics is required") + } + + base := channels.NewBaseChannel("mqtt", cfg, messageBus, cfg.AllowFrom, + channels.WithMaxMessageLength(4000), + channels.WithGroupTrigger(cfg.GroupTrigger), + channels.WithReasoningChannelID(cfg.ReasoningChannelID), + ) + + return &MQTTChannel{ + BaseChannel: base, + config: cfg, + }, nil +} + +// Name returns the channel name. +func (c *MQTTChannel) Name() string { + return "mqtt" +} + +// Start connects to the MQTT broker and begins listening. +func (c *MQTTChannel) Start(ctx context.Context) error { + logger.InfoC("mqtt", "Starting MQTT channel") + + c.ctx, c.cancel = context.WithCancel(ctx) + + opts := mqtt.NewClientOptions() + opts.AddBroker(c.config.Broker) + opts.SetClientID(c.config.ClientID) + opts.SetUsername(c.config.Username) + opts.SetPassword(c.config.Password) + opts.SetCleanSession(true) + opts.SetAutoReconnect(true) + opts.SetConnectRetry(true) + opts.SetConnectRetryInterval(5 * time.Second) + opts.SetMaxReconnectInterval(5 * time.Minute) + opts.SetKeepAlive(60 * time.Second) + opts.SetPingTimeout(10 * time.Second) + opts.SetWriteTimeout(10 * time.Second) + opts.SetConnectTimeout(30 * time.Second) + + // TLS configuration + if c.config.TLS { + tlsConfig := &tls.Config{ + InsecureSkipVerify: false, + } + if c.config.TLSCA != "" { + // Load CA cert if provided + // For simplicity, assuming file path + // In production, load cert properly + } + if c.config.TLSCert != "" && c.config.TLSKey != "" { + // Load client cert + } + opts.SetTLSConfig(tlsConfig) + } + + // Set message handler + opts.SetDefaultPublishHandler(c.onMessage) + + c.client = mqtt.NewClient(opts) + + // Connect with retry + if token := c.client.Connect(); token.Wait() && token.Error() != nil { + return fmt.Errorf("mqtt connect failed: %w", token.Error()) + } + + logger.InfoCF("mqtt", "Connected to MQTT broker", map[string]any{ + "broker": c.config.Broker, + "client_id": c.config.ClientID, + }) + + // Subscribe to topics + for _, topic := range c.config.SubscribeTopics { + if token := c.client.Subscribe(topic, byte(c.config.QoS), nil); token.Wait() && token.Error() != nil { + logger.ErrorCF("mqtt", "Failed to subscribe to topic", map[string]any{ + "topic": topic, + "error": token.Error(), + }) + } else { + logger.InfoCF("mqtt", "Subscribed to topic", map[string]any{ + "topic": topic, + }) + } + } + + c.SetRunning(true) + logger.InfoC("mqtt", "MQTT channel started") + return nil +} + +// Stop disconnects from the MQTT broker. +func (c *MQTTChannel) Stop(ctx context.Context) error { + logger.InfoC("mqtt", "Stopping MQTT channel") + c.SetRunning(false) + + if c.cancel != nil { + c.cancel() + } + + if c.client != nil && c.client.IsConnected() { + c.client.Disconnect(250) + } + + logger.InfoC("mqtt", "MQTT channel stopped") + return nil +} + +// onMessage handles incoming MQTT messages. +func (c *MQTTChannel) onMessage(client mqtt.Client, msg mqtt.Message) { + logger.DebugCF("mqtt", "Received message", map[string]any{ + "topic": msg.Topic(), + "payload": string(msg.Payload()), + }) + + // Try to parse as JSON first + var mqttMsg MQTTMessage + var content string + var err error + + // First try to parse as JSON + if err = json.Unmarshal(msg.Payload(), &mqttMsg); err == nil { + // Successfully parsed as JSON + content = mqttMsg.Status + } else { + // If JSON parsing fails, try to clean up common malformed JSON issues + payloadStr := string(msg.Payload()) + + // Try to extract JSON from malformed strings (e.g., extra quotes or braces) + // Look for a valid JSON object within the string + if strings.HasPrefix(payloadStr, "{") && (strings.HasSuffix(payloadStr, "}") || strings.HasSuffix(payloadStr, "}\"")) { + // Try to parse as-is first + if err2 := json.Unmarshal(msg.Payload(), &mqttMsg); err2 == nil { + content = mqttMsg.Status + } else { + // Try to clean up common issues + cleaned := strings.TrimSpace(payloadStr) + + // Remove extra quotes and braces from the end + for strings.HasSuffix(cleaned, "}") && strings.Count(cleaned, "{") < strings.Count(cleaned, "}") { + cleaned = cleaned[:len(cleaned)-1] + } + for strings.HasSuffix(cleaned, "\"}") && strings.Count(cleaned, "{") < strings.Count(cleaned, "}") { + cleaned = cleaned[:len(cleaned)-1] + } + // Remove any trailing quotes (simple check for extra quotes at the end) + for strings.HasSuffix(cleaned, "\"") && !strings.HasSuffix(cleaned, "\"}") { + cleaned = cleaned[:len(cleaned)-1] + } + + // Remove extra opening braces + for strings.HasPrefix(cleaned, "{") && strings.Count(cleaned, "{") > strings.Count(cleaned, "}") { + cleaned = cleaned[1:] + } + + if cleaned != payloadStr { + if err3 := json.Unmarshal([]byte(cleaned), &mqttMsg); err3 == nil { + content = mqttMsg.Status + logger.InfoCF("mqtt", "Successfully parsed cleaned JSON", map[string]any{ + "original": payloadStr, + "cleaned": cleaned, + }) + } else { + // Fall back to plain text + content = payloadStr + logger.InfoCF("mqtt", "Received plain text message (JSON parsing failed)", map[string]any{ + "error": err.Error(), + "payload": content, + }) + } + } else { + // Fall back to plain text + content = payloadStr + logger.InfoCF("mqtt", "Received plain text message (JSON parsing failed)", map[string]any{ + "error": err.Error(), + "payload": content, + }) + } + } + } else { + // Not JSON-like, treat as plain text + content = payloadStr + logger.InfoCF("mqtt", "Received plain text message (not JSON-like)", map[string]any{ + "payload": content, + }) + } + } + + if content == "" { + logger.WarnC("mqtt", "Empty content in MQTT message") + return + } + + // Determine sender ID + senderID := fmt.Sprintf("mqtt:%s", c.config.ClientID) + if msg.Topic() != "" { + senderID = fmt.Sprintf("mqtt:%s", strings.ReplaceAll(msg.Topic(), "/", "_")) + } + + // Check if message has reply-to header (in payload or topic) + replyTopic := c.config.ReplyTopic + if strings.Contains(content, "reply-to:") { + // Simple parsing for reply-to + parts := strings.SplitN(content, "reply-to:", 2) + if len(parts) == 2 { + replyTopic = strings.TrimSpace(parts[1]) + content = strings.TrimSpace(parts[0]) + } + } + + // Create peer and sender info + peer := bus.Peer{Kind: "direct", ID: msg.Topic()} // MQTT topics are like channels + sender := bus.SenderInfo{ + Platform: "mqtt", + PlatformID: senderID, + CanonicalID: senderID, + Username: senderID, + DisplayName: senderID, + } + + messageID := fmt.Sprintf("mqtt-%d", time.Now().UnixNano()) + + metadata := map[string]string{ + "platform": "mqtt", + "topic": msg.Topic(), + "reply_topic": replyTopic, + } + + if c.config.Instruction != "" { + content = c.config.Instruction + "\n\n" + content + } + + c.HandleMessage(c.ctx, peer, messageID, senderID, msg.Topic(), content, nil, metadata, sender) +} + +// Send sends a message to an MQTT topic. +func (c *MQTTChannel) Send(ctx context.Context, msg bus.OutboundMessage) error { + if !c.IsRunning() { + return fmt.Errorf("MQTT channel is not running") + } + + replyTopic := c.config.ReplyTopic + + // If chatID contains reply topic info, use it + if strings.HasPrefix(msg.ChatID, "reply:") { + replyTopic = strings.TrimPrefix(msg.ChatID, "reply:") + } + + // Replace placeholders in reply topic + replyTopic = strings.ReplaceAll(replyTopic, "{client_id}", c.config.ClientID) + replyTopic = strings.ReplaceAll(replyTopic, "{topic}", msg.ChatID) + // Add more placeholders as needed + + mqttMsg := MQTTMessage{ + Status: msg.Content, + } + + payload, err := json.Marshal(mqttMsg) + if err != nil { + return fmt.Errorf("failed to marshal MQTT message: %w", err) + } + + token := c.client.Publish(replyTopic, byte(c.config.QoS), c.config.Retain, payload) + if token.Wait() && token.Error() != nil { + return fmt.Errorf("failed to publish MQTT message: %w", token.Error()) + } + + logger.DebugCF("mqtt", "Published message", map[string]any{ + "topic": replyTopic, + "payload": string(payload), + }) + return nil +} \ No newline at end of file diff --git a/pkg/config/config.go b/pkg/config/config.go index 070e8e499..95b827a91 100644 --- a/pkg/config/config.go +++ b/pkg/config/config.go @@ -340,6 +340,7 @@ type ChannelsConfig struct { Pico PicoConfig `json:"pico"` PicoClient PicoClientConfig `json:"pico_client"` IRC IRCConfig `json:"irc"` + MQTT MQTTConfig `json:"mqtt"` } // GroupTriggerConfig controls when the bot responds in group chats. @@ -592,6 +593,27 @@ type IRCConfig struct { ReasoningChannelID string `json:"reasoning_channel_id" env:"PICOCLAW_CHANNELS_IRC_REASONING_CHANNEL_ID"` } +type MQTTConfig struct { + Enabled bool `json:"enabled" env:"PICOCLAW_CHANNELS_MQTT_ENABLED"` + Broker string `json:"broker" env:"PICOCLAW_CHANNELS_MQTT_BROKER"` + ClientID string `json:"client_id" env:"PICOCLAW_CHANNELS_MQTT_CLIENT_ID"` + Username string `json:"username" env:"PICOCLAW_CHANNELS_MQTT_USERNAME"` + Password string `json:"password" env:"PICOCLAW_CHANNELS_MQTT_PASSWORD"` + SubscribeTopics []string `json:"subscribe_topics" env:"PICOCLAW_CHANNELS_MQTT_SUBSCRIBE_TOPICS"` + ReplyTopic string `json:"reply_topic" env:"PICOCLAW_CHANNELS_MQTT_REPLY_TOPIC"` + TLS bool `json:"tls" env:"PICOCLAW_CHANNELS_MQTT_TLS"` + TLSCA string `json:"tls_ca" env:"PICOCLAW_CHANNELS_MQTT_TLS_CA"` + TLSCert string `json:"tls_cert" env:"PICOCLAW_CHANNELS_MQTT_TLS_CERT"` + TLSKey string `json:"tls_key" env:"PICOCLAW_CHANNELS_MQTT_TLS_KEY"` + QoS int `json:"qos" env:"PICOCLAW_CHANNELS_MQTT_QOS"` + Retain bool `json:"retain" env:"PICOCLAW_CHANNELS_MQTT_RETAIN"` + Prefix string `json:"prefix" env:"PICOCLAW_CHANNELS_MQTT_PREFIX"` + Instruction string `json:"instruction" env:"PICOCLAW_CHANNELS_MQTT_INSTRUCTION"` + AllowFrom FlexibleStringSlice `json:"allow_from" env:"PICOCLAW_CHANNELS_MQTT_ALLOW_FROM"` + GroupTrigger GroupTriggerConfig `json:"group_trigger,omitempty"` + ReasoningChannelID string `json:"reasoning_channel_id" env:"PICOCLAW_CHANNELS_MQTT_REASONING_CHANNEL_ID"` +} + type HeartbeatConfig struct { Enabled bool `json:"enabled" env:"PICOCLAW_HEARTBEAT_ENABLED"` Interval int `json:"interval" env:"PICOCLAW_HEARTBEAT_INTERVAL"` // minutes, min 5 diff --git a/pkg/gateway/gateway.go b/pkg/gateway/gateway.go index 7f920a6f1..8a3db3fee 100644 --- a/pkg/gateway/gateway.go +++ b/pkg/gateway/gateway.go @@ -21,6 +21,7 @@ import ( _ "github.com/sipeed/picoclaw/pkg/channels/line" _ "github.com/sipeed/picoclaw/pkg/channels/maixcam" _ "github.com/sipeed/picoclaw/pkg/channels/matrix" + _ "github.com/sipeed/picoclaw/pkg/channels/mqtt" _ "github.com/sipeed/picoclaw/pkg/channels/onebot" _ "github.com/sipeed/picoclaw/pkg/channels/pico" _ "github.com/sipeed/picoclaw/pkg/channels/qq"